使用Python实现单消费者、多生产者场景的最佳方式是什么？
使用Python实现单消费者、多生产者场景的最佳方式是什么？
提问于 2015-06-11 05:43:04
我有一个Python程序，它产生多个生产者线程，然后有一个循环等待Queue对象在其中有一些东西。它看起来像这样
for t in threads:
t.start()
while len(threads):
if not queue.empty():
response = queue.get()
# handle response
queue.task_done()
else:
sleep(1)
threads = [t for t in threads if t.is_alive()]必须有一种更优雅的方式来做到这一点。我已经研究了threading模块提供的所有同步对象，但我不知道如何应用它们。
仅供参考，我的代码可以满足我想要做的事情。我坚信不要修复没有损坏的东西，但我只是觉得有一种更好的方法可以做到这一点，一个更好的程序员会在第一时间做到这一点。
回答 2
Stack Overflow用户
发布于 2015-06-11 06:04:24
您可以使用weakref来测试线程是否仍处于活动状态：
import weakref
def consumer(queue, threads):
while threads:
try:
response = queue.get(timeout=1)
# handle response
queue.task_done()
except Empty:
pass
threads = weakref.WeakSet()
for i in range(10):
t = threading.Thread(...)
t.start()
threads.add(t)
del t # remove all references to threads
consumer(queue, threads)Stack Overflow用户
发布于 2015-06-11 07:02:58
@ Daniel :虚弱裁判是一个很酷的技巧。这是另一种方法，它只使用添加了“终止策略”的队列。
您将需要确保每个生产者的线程目标函数总是将最终的“终止消息”放到队列中，本质上是在它们完成生成后为“无”。消费者只需等待，直到接收到适当数量的终止(每个生产者线程1个)，然后退出循环。这样，您就不必检查线程是否已经结束，实际上只有一个通信点:队列。但是，如果使用者中有异常，生产者线程可能会处于“守护进程”模式，这样它们就不会在等待使用者队列时阻塞进程。嗯，消耗掉了。
您必须确保终止消息总是以某种try-finally缩进的形式发送给每个生产者。否则，您将不得不在消费者的except中处理超时。
import functools
def consumer(queue,num_threads_remaining):
next_message=functools.partial(iter,functools.partial(queue.get,timeout=1),None)
while num_threads_remaining:
try:
for response in next_message():
# handle response.. hopefully exception-protected
queue.task_done()
# we got a None termination message
num_threads_remaining -= 1
except Empty: pass # handle some other check when idling?页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/30767828
复制相关文章
相似问题
腾讯云开发者
