一旦队列可用，就逐步提取物品答案

【问题标题】：Picking up items progressivly as soon as a queue is available一旦队列可用，就逐步提取物品
【发布时间】：2013-01-22 13:37:39
【问题描述】：

我正在寻找一个可靠的实现，以允许我使用 Queue 逐步处理项目列表。

我的想法是我想使用一定数量的工作人员，这些工作人员将完成 20 多个数据库密集型任务的列表并返回结果。我希望 Python 从前五个项目开始，一旦完成一个任务，就开始队列中的下一个任务。

这就是我目前在没有Threading 的情况下这样做的方式。

for key, v in self.sources.iteritems():
    # Do Stuff

我希望采用类似的方法，但可能不必将列表分成五个子组。这样它就会自动选择列表中的下一个项目。目标是确保如果一个数据库减慢了进程，它不会对整个应用程序产生负面影响。

【问题讨论】：

标签： python multithreading python-2.7 queue

【解决方案1】：

您可以自己实现这一点，但 Python 3 已经附带了一个基于 Executor 的线程管理解决方案，您可以通过安装 the backported version 在 Python 2.x 中使用它。

你的代码可能看起来像

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    future_to_key = {}
    for key, value in sources.items():
        future_to_idday[executor.submit(do_stuff, value)] = key
    for future in concurrent.futures.as_completed(future_to_key):
        key = future_to_key[future]
        result = future.result()
        # process result

【讨论】：

谢谢。我会试试看。忘了说我用的是 2.x。

【解决方案2】：

如果你使用的是 python3，我推荐并发期货模块。如果您没有使用 python3 并且没有附加到线程（相对于进程），那么您可以尝试 multiprocessing.Pool （尽管它带有一些警告，并且我在我的应用程序中无法正确关闭池时遇到了麻烦）。如果您必须使用线程，在 python2 中，您最终可能会自己编写代码 - 生成 5 个运行消费者函数的线程，然后简单地将调用（函数 + args）迭代地推送到队列中，以便消费者找到并处理它们。

【讨论】：

multiprrocessing.dummy also provides the same interface 使用线程而不是进程

【解决方案3】：

你可以只使用 stdlib 来做到这一点：

#!/usr/bin/env python
from multiprocessing.dummy import Pool # use threads

def db_task(key_value):
    try:
        key, value = key_value
        # compute result..
        return result, None
    except Exception as e:
        return None, e

def main():
    pool = Pool(5)
    for result, error in pool.imap_unordered(db_task, sources.items()):
        if error is None:
            print(result)

if __name__=="__main__":
    main()

【讨论】：