producer-consumer.py

from threading import Thread
from queue import Queue
from time import sleep


queue = Queue(maxsize=4)
EXIT_FLAG = object()


def producer():
    for i in range(10):
        print(f'Adding {i} to the queue..')
        queue.put(i)
    print('Producer done.')
    queue.put(EXIT_FLAG)


def consumer(id):
    while True:
        data = queue.get()
        if data == EXIT_FLAG:
            print(f'  Consumer {id} done.')
            # The remaining consumers will never exit if you comment this line.
            queue.put(EXIT_FLAG)
            break
        print(f'  [{id}] Consuming {data}..')
        # Simulate time-consuming operation
        sleep(.2)


Thread(target=producer).start()
threads = []
for i in range(3):
    t = Thread(target=consumer, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()
assert(queue.qsize() == 1)

queue.get()是阻塞的，不需要再sleep啊

@zh-h 只有 28 行有一个 sleep，是用来模拟处理数据时消耗的时间的。