queue用来在进程间传递消息，任何可以pickle-able的对象都可以在加入到queue。 multiprocessing.JoinableQueue 是 Queue的子类，增加了task_done()和join()方法。 task_done()用来告诉queue一个task完成。一般地在调用get()获得一个task，在task结束后调用task_done()来通知Queue当前task完成。 join() 阻塞直到queue中的所有的task都被处理（即task_done方法被调用）
import time import multiprocessing class Consumer(multiprocessing.Process): def __init__(self,task_queue,result_queue): multiprocessing.Process.__init__(self) self.task_queue = task_queue self.result_queue = result_queue def run(self): process_name = self.name while True: next_queue = self.task_queue.get() if next_queue is None: self.task_queue.task_done() print("%s,exiting"%(process_name)) break print("%s:%s"%(process_name,next_queue)) self.result_queue.put(next_queue()) self.task_queue.task_done() return class Task(object): def __init__(self,a,b): self.a = a self.b = b def __call__(self): # return "%s * %s = %s"%(self.a,self.b,self.a*self.b) pass def __str__(self): return "%s * %s"%(self.a,self.b) if __name__ == '__main__': # 1.创建队列 tasks = multiprocessing.JoinableQueue() results = multiprocessing.SimpleQueue() # 2.开启消费者 consumers = [Consumer(tasks,results) for i in range(20)] for t in consumers: t.start() # 3.任务队列 number_jobs = 10 for i in range(number_jobs): tasks.put(Task(i,i)) # 4. 添加终止None for i in range(number_jobs): tasks.put(None) tasks.join() # 5. 输出返回值 while number_jobs: result = results.get() print("result:",result) number_jobs -= 1
Consumer-42:1 * 1 Consumer-43:2 * 2 Consumer-44:3 * 3 Consumer-45:4 * 4 Consumer-41:0 * 0 Consumer-46:5 * 5 Consumer-47:6 * 6 Consumer-48:7 * 7 Consumer-49:8 * 8 Consumer-50:9 * 9 Consumer-53,exiting Consumer-52,exiting Consumer-54,exiting Consumer-51,exiting Consumer-60,exiting result: 1 * 1 = 1 result: 2 * 2 = 4 result: 3 * 3 = 9 result: 4 * 4 = 16 result: 5 * 5 = 25 result: 0 * 0 = 0 result: 6 * 6 = 36 result: 7 * 7 = 49 result: 8 * 8 = 64 result: 9 * 9 = 81 Consumer-58,exiting Consumer-55,exiting Consumer-56,exiting Consumer-57,exiting Consumer-59,exiting