python 线程通信 生产者与消费者

Posted 2020-12-07 乘风去破浪

  1 """
  2 线程通信的生产者与消费者
  3 python的queue模块中提供了同步的线程安全的队列类,都具有原子性，实现线程间的同步
  4 Queue （FIFO： fist in fist out）
  5 LifoQueue （LIFO： last in fist out）
  6 PriorityQueue (优先级队列)
  7 
  8 task_done()：
  9 作用是在使用join()的时候，当queue中所有的项目都被取出，且每个项目取出后都使用了task_done()，那么就可以释放join()阻塞
 10 系统解释如下：
 11 用于消费者，每次get()以后，使用task_done() 是告诉队列正在处理的get任务完成
 12 如果join()当前处于阻塞状态，那么当处理完所有项时它将继续运行(这意味着对于已经放入队列的每个项都接收到task_done()调用)。
 13 如果调用的次数超过在队列中放置的项的次数，则引发ValueError错误。
 14 
 15 如果不需要join()的时候也可以不使用task_done()
 16 
 17 
 18 """
 19 import queue
 20 import threading
 21 import time
 22 import random
 23 
 24 q = queue.Queue(10)
 25 
 26 
 27 def produce():
 28     i = 0
 29     while i < 10:
 30         num = random.randint(1, 100)
 31         q.put("生产者生产出数据：%d" % num)
 32         print("生产者生产出数据：%d" % num)
 33         time.sleep(0.2)
 34         i += 1
 35     print("生产结束")
 36 
 37 
 38 def consume():
 39     while True:
 40         time.sleep(0.3)
 41         if q.empty():
 42             break
 43         item = q.get()
 44         print("消费者取出:", item)
 45         q.task_done()
 46 
 47     print("消费者结束")
 48 
 49 
 50 if __name__ == ‘__main__‘:
 51 
 52     # 创建生产者
 53     t1 = threading.Thread(target=produce, name="生产者")
 54     t1.start()
 55     time.sleep(0.1)
 56 
 57     # 创建消费者
 58     t2 = threading.Thread(target=consume, name="消费者")
 59     t2.start()
 60     q.join()
 61 
 62     print("over")
 63 
 64 
 65 # from threading import Thread
 66 # import time
 67 # import random
 68 # from queue import Queue
 69 # from collections import deque
 70 # from datetime import datetime
 71 #
 72 # # 创建队列，设置队列最大数限制为3个
 73 # queue = Queue(3)
 74 #
 75 #
 76 # # 生产者线程
 77 # class Pro_Thread(Thread):
 78 #     def run(self):
 79 #         # 原材料准备，等待被生产,这里使用的是双向队列
 80 #         tasks = deque([1, 2, 3, 4, 5, 6, 7, 8])
 81 #         global queue
 82 #         while True:
 83 #             try:
 84 #                 # 从原材料左边开始生产，如果tasks中没有元素，调用popleft()则会抛出错误
 85 #                 task = tasks.popleft()
 86 #                 queue.put(task)
 87 #                 print(datetime.now(), "生产", task, "现在队列数:", queue.qsize())
 88 #
 89 #                 # 休眠随机时间
 90 #                 time.sleep(0.5)
 91 #             # 如果原材料被生产完，生产线程跳出循环
 92 #             except IndexError:
 93 #                 print("原材料已被生产完毕")
 94 #                 break
 95 #         print("生产完毕")
 96 #
 97 #
 98 # # 消费者线程
 99 # class Con_Thread(Thread):
100 #     def run(self):
101 #         global queue
102 #         while True:
103 #             if not queue.empty():
104 #                 # 通过get(),这里已经将队列减去了1
105 #                 task = queue.get()
106 #                 time.sleep(2)
107 #                 # 发出完成的信号，不发的话，join会永远阻塞，程序不会停止
108 #                 queue.task_done()
109 #                 print(datetime.now(), "消费", task)
110 #             else:
111 #                 break
112 #         print("消费完毕")
113 #
114 #
115 # # r入口方法，主线程
116 # def main():
117 #     Pro_1 = Pro_Thread()
118 #     # 启动线程
119 #     Pro_1.start()
120 #     # 这里休眠一秒钟，等到队列有值，否则队列创建时是空的，主线程直接就结束了，实验失败，造成误导
121 #     time.sleep(1)
122 #     for i in range(2):
123 #         Con_i = Con_Thread()
124 #         # 启动线程
125 #         Con_i.start()
126 #     global queue
127 #     # 接收信号，主线程在这里等待队列被处理完毕后再做下一步
128 #     queue.join()
129 #     # 给个标示，表示主线程已经结束
130 #     print("主线程结束")
131 #
132 #
133 # if __name__ == ‘__main__‘:
134 #     main()