1

我有一个 python 脚本，在我的文件顶部：

result_queue = Queue.Queue()
key_list = *a large list of small items* #(actually from bucket.list() via boto)

我了解到队列是过程安全的数据结构。我有一个方法：

def enqueue_tasks(keys):
    for key in keys:
        try:
            result = perform_scan.delay(key)
            result_queue.put(result)
        except:
           print "failed"

这里的perform_scan.delay()函数实际上调用了一个 celery worker，但我认为不相关（它是一个异步进程调用）。

我也有：

def grouper(iterable, n, fillvalue=None):
    args = [iter(iterable)] * n
    return izip_longest(fillvalue=fillvalue, *args)

最后我有一个main()功能：

def main():

    executor = concurrent.futures.ProcessPoolExecutor(10)
    futures = [executor.submit(enqueue_tasks, group) for group in grouper(key_list, 40)]
    concurrent.futures.wait(futures)
    print len(result_queue)

print 语句的结果是 0。但是，如果我包含一个大小为result_queuein的 print 语句enqueue_tasks，当程序运行时，我可以看到大小正在增加，并且正在将内容添加到队列中。

对正在发生的事情的想法？

4

2 回答 2

2

您需要使用 a multiprocessing.Queue，而不是 a Queue.Queue。Queue.Queue是线程安全的，而不是进程安全的，因此您在一个进程中对其所做的更改不会反映在任何其他进程中。

于 2014-10-17T00:00:30.327 回答
2

看起来这个问题有一个更简单的解决方案。

你正在建立一个期货清单。期货的全部意义在于它们是未来的结果。特别是，无论每个函数返回什么，这都是未来的（最终）价值。所以，根本不要做整个“将结果推送到队列”的事​​情，只需从任务函数中返回它们，然后从期货中提取它们。

最简单的方法是打破这个循环，让每个键都是一个单独的任务，有一个单独的未来。我不知道这是否适合您的真实代码，但如果是：

def do_task(key):
    try:
        return perform_scan.delay(key)
    except:
        print "failed"

def main():
    executor = concurrent.futures.ProcessPoolExecutor(10)
    futures = [executor.submit(do_task, key) for key in key_list]
    # If you want to do anything with these results, you probably want
    # a loop around concurrent.futures.as_completed or similar here,
    # rather than waiting for them all to finish, ignoring the results,
    # and printing the number of them.
    concurrent.futures.wait(futures)
    print len(futures)

当然，这不会进行分组。但是你需要吗？

需要分组的最可能原因是任务非常小，以至于调度它们（以及对输入和输出进行腌制）的开销淹没了实际工作。如果这是真的，那么您几乎可以肯定地等到整个批次完成后才能返回任何结果。尤其是考虑到在它们全部完成之前你甚至都不看结果。（这种“分成组，处理每个组，重新合并在一起”的模型在数值工作等情况下非常常见，其中每个元素可能很小，或者元素可能不相互独立，但有些组很大足够或独立于其余工作。）

无论如何，这几乎一样简单：

def do_tasks(keys):
    results = []
    for key in keys:
        try:
            result = perform_scan.delay(key)
            results.append(result)
        except:
           print "failed"
    return results

def main():
    executor = concurrent.futures.ProcessPoolExecutor(10)
    futures = [executor.submit(enqueue_tasks, group) for group in grouper(key_list, 40)]
    print sum(len(results) for results in concurrent.futures.as_completed(futures))

或者，如果您更喜欢先等待然后计算：

def main():
    executor = concurrent.futures.ProcessPoolExecutor(10)
    futures = [executor.submit(enqueue_tasks, group) for group in grouper(key_list, 40)]
    concurrent.futures.wait(futures)
    print sum(len(future.result()) for future in futures)

但同样，我怀疑你甚至需要这个。

于 2014-10-17T00:16:56.237 回答