使用python的threading中的Thread
下面是两种基本的实现线程的方式：
第一种方式————
#coding=utf-8 """
thread的第一种声明及调用方式
""" from threading import Thread
import time def func():
while True:
print time.ctime()
time.sleep(1) def func1():
while True:
print "func1", time.ctime()
time.sleep(2) #将要执行的方法作为参数传给Thread的构造方法 t = Thread(target=func)
t.start() t1 = Thread(target=func1)
t1.start()
第二种方式————
#coding=utf-8 """
thread的第二种声明及调用方法
""" import time
from threading import Thread
from time import sleep #从Thread继承，并重写run()
class myThread(Thread):
def __init__(self, threadname):
Thread.__init__(self, name=threadname) def run(self):
while True:
print self.getName() + " " + time.ctime()
time.sleep(1) t = myThread("thread_1")
t.start() t1 = myThread("thread_2")
t1.start()
接下来是：join和setDaemon的方法调用及具体区别：
join的方法调用：
#coding=utf-8 """
join方法的含义是：存在A主线程，B子线程，在A主线程中生成了一个子线程B，
如果子线程B调用了join方法，则主线程需要等待子线程操作结束，才能够继续往下执行，主线程会被阻塞
"""
import time
from threading import Thread class myThread(Thread):
def __init__(self, threadname):
Thread.__init__(self, name = threadname) def run(self):
time.sleep(10)
print self.getName() + " " + time.ctime() def mainfunc():
for i in range(5):
print i if __name__ == "__main__":
t = myThread("childthread")
t.start()
#增加join方法的调用，则程序运行时的表现是：先子线程（会先sleep 10s，之后打印子线程的内容），再运行主线程中的打印从0到4的逻辑
#不增加join方法的调用，则程序运行时的表现是：会先走主线程的打印从0到4的逻辑，之后子线程（sleep 10s，再打印子线程中的内容），但其实这里就是没有等待子线程
# t.join()
mainfunc()
setDaemon方法的调用：
#coding=utf-8 """
daemon方法的含义是：存在主线程A，和子线程B，子线程B是从主线程A中生成的
如果设置子线程B为daemon即守护进程，则主线程A不存在的话，子线程B就退出
如果没有设置子线程B为daemon，则主线程A结束运行退出之后，子线程B也不退出
""" import time
from threading import Thread class myThread(Thread):
def __init__(self, threadname):
Thread.__init__(self, name=threadname) def run(self):
time.sleep(10)
print self.getName() + " " + time.ctime() def mainfunc():
for i in range(5):
print i if __name__ == "__main__":
t = myThread("zixiancheng")
#不设置子线程为守护进程，则主线程退出，子线程继续运行
#设置子线程为守护进程，则主线程退出，子线程直接退出
#并且，需要在start之前调用setDaemon
# t.setDaemon(True)
t.start()
mainfunc()
接下来是线程同步相关的内容：
首先是没有线程同步机制的情况下，多个线程同时操作一个公共数据：
代码如下：
#coding=utf-8 """
@function: test Thread的Lock
@author: LiXia
@date: 2017-0303
""" """
所有的锁都是为了在线程之间进行通信
例如对一个公共变量进行操作，此例子为没有使用线程同步机制对公共变量进行操作
""" from threading import Thread
import time data = 0 class myThread(Thread):
def __init__(self):
Thread.__init__(self) def run(self):
global data
time.sleep(1)
data += 1
print self.getName() + " data: " + str(data) if __name__ == "__main__":
for i in range(5):
t = myThread()
t.start()
运行结果见下方：
为了能够对公共操作的数据，在特定逻辑内保持一致，需要保证在该逻辑内，只有单个线程可以操作，其他线程需要等待该线程处理结束才可以。
所以下面是关于python线程同步机制提供的指令：
1、Lock：
但是将以上代码中增加上lock处理，注意一定要增加在对公共数据的逻辑判断和处理的前面和后面，所以这里必须添加到while判断之前，释放锁需要在输出操作结束之后添加
即：
#coding=utf-8 """
所有的锁都是为了在线程之间进行通信
例如对一个公共变量进行操作
""" from threading import Thread
import threading
import time data = 0
lock = threading.Lock() class myThread(Thread):
def __init__(self):
Thread.__init__(self) def run(self):
global data
if lock.acquire():
time.sleep(1)
data += 1
print self.getName() + " data: " + str(data)
lock.release() if __name__ == "__main__":
for i in range(5):
t = myThread()
t.start()
运行结果见下方：
2、RLock
代码如下：
#coding=utf-8 """
通过RLock（可重入锁）进行线程间的安全通信,RLock为可重入锁，即线程在拥有RLock之后可以继续acquire()，只要保证acquire()次数与
release()次数一致即可；
继续对公共变量进行操作
""" from threading import Thread
import threading
import time rlock = threading.RLock()
data = 0 class myThread(Thread):
def __init__(self):
Thread.__init__(self) def run(self):
global data
if rlock.acquire():
time.sleep(1)
data += 1
print self.getName() + " data:" + str(data) if rlock.acquire():
time.sleep(1)
data -= 1
print self.getName() + " data:" + str(data)
rlock.release()
rlock.release() if __name__ == "__main__":
for i in range(5):
t = myThread()
t.start()
运行结果：
3、Condition
Condition就是条件变量，condition通常与一个锁关联。
acquire([timeout])/release(): 调用关联的锁的相应方法。
wait([timeout]): 调用这个方法将使线程进入Condition的等待池等待通知，并释放锁。使用前线程必须已获得锁定，否则将抛出异常。
notify(): 调用这个方法将从等待池挑选一个线程并通知，收到通知的线程将自动调用acquire()尝试获得锁定（进入锁定池）；其他线程仍然在等待池中。调用这个方法不会释放锁定。使用前线程必须已获得锁定，否则将抛出异常。
notifyAll(): 调用这个方法将通知等待池中所有的线程，这些线程都将进入锁定池尝试获得锁定。调用这个方法不会释放锁定。使用前线程必须已获得锁定，否则将抛出异常。
#coding=utf-8 """
以最经典的生产者和消费者的问题为例，两者需要针对一个公共的数据区域做操作，并且两端需要进行通信
生产者在生产之前，需要先占领地盘，然后开始生产，生产结束通知消费者来取；
消费者在消费的时候，也需要先占领地盘，然后开始消费，消费结束通知生产者继续生产；
当然还有多个生产者和消费者的情况，会更加复杂，可以用python的Queue来实现
""" from threading import Thread
import threading
import time
import random product = []
con = threading.Condition() #得到一个Condition的对象 #生产者线程
class producerThread(Thread):
def __init__(self, threadname):
Thread.__init__(self, name=threadname) def run(self):
global product
if con.acquire():
while True:
r = random.randint(0, 20)
product.append(r)
print "product len and product: ", self.getName(), product
con.notify()
con.wait()
# time.sleep(2) #消费者线程
class consumerThread(Thread):
def __init__(self, threadname):
Thread.__init__(self, name=threadname) def run(self):
global product
if con.acquire():
while True:
if product != []:
product.pop()
print "consumer len and product: ", self.getName(), product
con.notify()
con.wait()
time.sleep(2) if __name__ == "__main__":
t1 = producerThread("producer")
t2 = consumerThread("consumer") t1.start()
t2.start()
运行结果如下：
4、对多个生产者消费者支持的很好的Queue
Queue是什么？Queue实现了多生产者-多消费者的队列形式，分别包含先入先出（默认）FIFO、后进先出LIFO、以及优先级队列PriorityQueue（优先级低的先出）
Queue模块的class及exception：
#coding=utf-8 """
Queue的类和异常
""" import Queue q = Queue.Queue(maxsize = 0) #FIFO的Queue，maxsize即Queue的size值，maxsize≤0表示无size限制 lq = Queue.LifoQueue(maxsize = 0) #LIFO的Queue，先进后出，maxsiz的含义同上 pq = Queue.PriorityQueue(maxsize = 0) #优先级Queue，优先级低的先出，maxsize的含义同上 Queue.Empty() #当Queue的对象为空时，使用非阻塞get时会抛出异常 Queue.Full() #当Queue的对象已满时，使用非阻塞put时会抛出异常
Queue的对象：
Queue（Queue、LifoQueue、PriorityQueue）的对象提供的公共方法如下：
#coding=utf-8 """
Queue（Queue、LifoQueue、PriorityQueue）的对象的公共方法
""" import Queue q = Queue.Queue() q.qsize() #获取Queue的大小
q.empty() #是否为空，是则返回True，否则返回False
q.full() #是否已满，是则返回True，否则返回False
q.put(item[, block[, timeout]]) #put元素到Queue中，可选参数block默认值为True,timeout默认值为None，即为阻塞方法，
#接上，若timeout为正值，则代表会阻塞timeout的时长之后引发一个full的异常（如果已满）
q.put_nowait(item) #相当于q.put(item, False)
q.get([block[, timeout]]) #get元素到Queue中，可选参数block默认值为True，timeout默认值为None，即为阻塞方法，
#接上，若timeout为正数，则代表会则色timeouu的时长之后会引发一个empty的异常（如果为空）
q.get_nowait() #相当于q.get(False)
q.task_done() #用在Queue的消费者模式下，即用在Queue.get()方法之后，通过Queue.task_done()告诉Queue当前任务完成
#如果当前join正在阻塞，则它将在所有item都被处理之后恢复
q.join() #会保持阻塞直到队列中的所有item都会处理
#接上，未完成的task数目会增加当一个item被add到Queue中，未完成的task数目会降低，每当消费者线程调用task_done来表示所有的
#接上，work都完成了，当未完成的数目降低到0时，join就自动取消阻塞了
Queue的主要用法是什么？
看一下Queue的源码实现，然后在另一篇博客中更新一下。
#coding=utf-8 """
理解Queue的用法，理解Queue衍生出来的不同的class
""" from Queue import Queue
from threading import Thread
import time
import random q = Queue() class producerQueue(Thread):
def __init__(self, queue):
Thread.__init__(self)
self.queue = queue def run(self):
while True:
r = random.randint(0, 99)
self.queue.put(r)
print self.getName() + "produce " + str(r)
time.sleep(1) class consumerQueue(Thread):
def __init__(self, queue):
Thread.__init__(self)
self.queue = queue def run(self):
while True:
r = self.queue.get()
print self.getName() + "consume " + str(r)
time.sleep(3)
self.queue.task_done() if __name__ == "__main__":
threads = []
for i in range(5):
t = producerQueue(q)
t.start()
# threads.append(t) for i in range(3):
t = consumerQueue(q)
t.start()
# threads.append(t) q.join()
之后得到的结果如下：