多线程通知
Python中threading.Event可以实现线程间的事件通知。
主要有以下方法：
wait() #-->等待事件者调用wait来等待事件，wait可以接受到参数，等待多少秒，默认是一直阻塞等待下去
set() #-->通知事件者调用set来通知事件
clear() #-->清除写事件，以方便下次读取
多线程通信
Queue模块用来实现消息队列功能，可以实现线程间安全的消息交换。各个线程可以通过调用消息队列实例对消息队列进行操纵。
queue对象实现一个fifo队列（其他的还有lifo、priority队列，这里不再介绍）。queue只有qsize一个构造参数，用来指定队列容量，指定为0的时候代表容量无限。主要有以下成员函数：
Queue.qsize()：返回消息队列的当前空间。返回的值不一定可靠。
Queue.empty()：判断消息队列是否为空，返回True或False。同样不可靠。
Queue.full()：类似上边，判断消息队列是否满
Queue.put(item, block=True, timeout=None)：往消息队列中存放消息。block可以控制是否阻塞，timeout指定阻塞时候的等待时间。如果不阻塞或者超市，会引起一个full exception。
Queue.put_nowait(item)：相当于put(item, False).
Queue.get(block=True, timeout=None)：获取一个消息，其他同put。
以下两个函数用来判断消息对应的任务是否完成。
Queue.task_done()：接受消息的线程通过调用这个函数来说明消息对应的任务已完成。
Queue.join()：调用线程阻塞直到所有消息对应的任务已经完成。
实现的边下载边计算的demo
# -*- coding: utf-8 -*-
import threading
import Queue
from threading import Thread
import time
class ThreadManager():
"""manage download compute and report
"""
def __init__(self):
# a queue for computing
self.compute_queue = Queue.Queue()
# a queue for downloading
self.download_queue = Queue.Queue()
# is downloading or not
self.download_event = threading.Event()
# is computing or not
self.compute_event = threading.Event()
# all download and compute tasks are done if true
self.report_event = threading.Event()
pass
def run(self):
CThread = ComputeThread(download_queue=self.download_queue,
compute_queue=self.compute_queue,
download_event=self.download_event,
compute_event=self.compute_event,
report_event=self.report_event)
DThread = DownloadThread(download_queue=self.download_queue,
compute_queue =self.compute_queue,
download_event=self.download_event,
compute_event=self.compute_event,
report_event=self.report_event)
CThread.start()
DThread.start()
while True:
for i in range(10):
self.download_queue.put(i)
self.download_event.set()
print "download event start"
self.report_event.wait()
print "all task done"
self.report_event.clear()
class ComputeThread(Thread):
def __init__(self,download_queue,compute_queue,download_event,compute_event,report_event):
Thread.__init__(self, name='compute')
self.download_queue = download_queue
self.compute_queue = compute_queue
self.download_event = download_event
self.compute_event = compute_event
self.report_event = report_event
def run(self):
while True:
self.compute_event.wait()
if self.compute_queue.qsize() > 0:
item = self.compute_queue.get()
print "[compute %d]"%item
time.sleep(1) # compute task
else:
self.compute_event.clear()
print "no data, compute wait"
if not self.download_event.isSet():
self.report_event.set()
print "-------------report----------------"
class DownloadThread(Thread):
def __init__(self,download_queue,compute_queue,download_event,compute_event,report_event):
Thread.__init__(self, name='download')
self.download_queue = download_queue
self.compute_queue = compute_queue
self.download_event = download_event
self.compute_event = compute_event
self.report_event = report_event
def run(self):
while True:
self.download_event.wait()
if self.download_queue.qsize() > 0:
item = self.download_queue.get()
print "[download %d]"%item
time.sleep(0.3) # download task
self.compute_queue.put(item)
self.compute_event.set()
else:
self.download_event.clear()
print "no data, download wait"
if __name__ == '__main__':
task = ThreadManager()
task.run()
运行截图如下所示
QQ20180404-171826@2x.png
