在 python 中的生产者和消费者线程上需要帮助

Question

我想同时在python中创建消费者和生产者线程，生产者线程将追加队列，消费者线程检索存储在队列中的项目。我需要与生产者一起启动消费者线程。消费者线程应该等到队列得到一个项目。当队列中没有项目时它应该终止。我是python新手，请帮忙。

要求：

如果有 10 个数字的列表，生产者线程应该插入一个包含一项的队列，消费者线程应该检索这个数字。两个线程应该同时启动。

from queue import Queue
import threading
import time

class producer(threading.Thread):

    def __init__(self, list_of_numbers):
        threading.Thread.__init__(self)
        self.list_items = list_of_numbers

    def run(self):
        for i in self.list_items:
            queue.put(str(i))

class consumer(threading.Thread):

    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        while queue.not_empty:
            queue_ret = queue.get()
            print("Retrieved", queue_ret)


queue = Queue()
producers = producer([10,20,5,4,3,2,1])
consumers = consumer()

producers.start()
consumers.start()
producers.join()
consumers.join()

Answer 1

完成后只需放置一个特殊物品：

_im_done = object()

class producer(threading.Thread):
    def run(self):
        '''feed the consumer until you are done'''
        queue.put(_im_done)

class consumer(threading.Thread):
    def run(self):
        while True:
            queue_ret = queue.get()
            if queue_ret is _im_done:
                break
            '''normal execution'''

如果有多个消费者，那么你必须在停止之前把物品放回去：

class consumer(threading.Thread):
    def run(self):
        while True:
            queue_ret = queue.get()
            if queue_ret is _im_done:
                queue.put(_im_done)
                break
            '''normal execution'''

Answer 2

您可以直接使用queue module。该文档包含您的用例的示例。附带说明一下，该模块在 Python 2 中被命名为 Queue。

但是，如果 Python 中的线程受 CPU 限制，它不会让您的程序更快，在这种情况下，您可以使用 multiprocessing module 代替（在 IO 限制的情况下，线程可能更可行，因为线程通常更便宜）。多处理模块还提供了一个名为multiprocessing.Queue 的安全队列实现。

Answer 3

queue.get() 正在阻塞。如果队列中没有项目，它将卡在那里。您应该使用while True: queue.get(block=False) 并处理空异常并退出。

确定完整代码以消除混乱

from Queue import Queue, Empty
import threading
import time

started = False
running = True

class producer(threading.Thread):

    def __init__(self, list_of_numbers):
        threading.Thread.__init__(self)
        self.list_items = list_of_numbers

    def run(self):
        started = True
        for i in self.list_items:
            queue.put(str(i))
        running = False

class consumer(threading.Thread):

    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        while not started:
            sleep(0)

        while running:
            try:
                queue_ret = queue.get(block=False)
            except Empty:
                sleep(0)
                continue
            print("Retrieved", queue_ret)


queue = Queue()
producers = producer([10,20,5,4,3,2,1])
consumers = consumer()

producers.start()
consumers.start()
producers.join()
consumers.join()