多处理队列最大大小限制为 32767答案

【问题标题】：Multiprocessing Queue maxsize limit is 32767多处理队列最大大小限制为 32767
【发布时间】：2011-08-19 13:10:31
【问题描述】：

我正在尝试使用多处理编写一个 Python 2.6 (OSX) 程序，并且我想用超过默认的 32767 个项目来填充队列。

from multiprocessing import Queue
Queue(2**15) # raises OSError

Queue(32767) 可以正常工作，但任何更大的数字（例如Queue(32768)）都会失败，OSError: [Errno 22] Invalid argument

这个问题有解决办法吗？

【问题讨论】：

队列包含什么样的数据？您确定任何更高的数字都会失败，还是第 32768 个数据节点会导致错误？（您是否偶然使用了路径名？）
@voithos，我没有在队列爆炸之前填充队列。仅设置最大大小会导致 OSError。

标签： python queue multiprocessing max-size

【解决方案1】：

我已经回答了最初的问题，但我确实想补充一点，Redis 列表非常可靠，Python 模块对它们的支持非常容易用于实现类似队列的对象。它们的优点是允许一个人在多个节点（跨网络）以及多个进程上进行横向扩展。

基本上要使用那些您只需为队列名称选择一个键（字符串）的方法，让您的生产者推入其中，并让您的工作人员（任务消费者）循环阻止来自该键的弹出。

Redis BLPOP 和 BRPOP 命令都采用一个键列表（列表/队列）和一个可选的超时值。他们返回一个元组（键，值）或无（超时）。因此，您可以轻松编写一个与熟悉的 select() 结构非常相似的事件驱动系统（但级别更高）。您唯一需要注意的是缺少键和无效的键类型（当然，只需使用异常处理程序包装您的队列操作）。（如果某些其他应用程序在您的共享 Redis 服务器上停止，删除键或用字符串/整数或其他类型的值替换您用作队列的键......那么，此时您会遇到不同的问题）。 :)

此模型的另一个优点是 Redis 确实将其数据保存到磁盘。因此，如果您选择允许，您的工作队列可以在系统重新启动后继续存在。

（当然，如果您真的想在 SQLlite 或任何其他 SQL 系统中将简单的 Queue 实现为表；只需使用某种自动递增索引进行排序，并使用列来标记每个项目已经“完成”（已使用）；但这确实比使用 Redis 提供的“开箱即用”更复杂）。

【讨论】：

【解决方案2】：

一种方法是使用自定义类包装您的multiprocessing.Queue（仅在生产者方面，或从消费者的角度透明地）。使用它，您可以将要分派到您正在包装的Queue 对象的项目排队，并且仅在空间可用时将本地队列（Python list() 对象）中的内容提供给multiprocess.Queue，并进行异常处理在Queue 已满时进行节流。

这可能是最简单的方法，因为它对您的其余代码的影响应该最小。自定义类的行为应该像队列一样，同时将底层 multiprocessing.Queue 隐藏在您的抽象后面。

（一种方法可能是让您的生产者使用线程，一个线程来管理从线程Queue 到您的multiprocessing.Queue 的调度，以及实际上只是提供线程Queue 的任何其他线程。

【讨论】：

这种方法的问题是（无论如何在 OS X 上），没有办法知道队列中有多少项目。调用.qsize 方法引发NotImplementedError，解释为“在Mac OSX 上引发NotImplementedError，因为sem_getvalue() 损坏”。同样，.full 方法在 OS X 上完全不可靠。因此，除非我弄错了，否则没有可靠的方法来获取实现您描述的包装类所需的信息......
@Peter McMahan：当您尝试 .put() 满时将其放入其中时，Multiprocessing.Queue 是否会引发合理的异常？请注意，我确实建议“使用异常处理”。
在我的 MacOS X (Lion 10.7.2) 上使用默认 Python 安装（版本 2.7.1）几分钟后，我发现它的 multiprocess.Queue 也有 32767 的限制（）容量。 .empty() 似乎和 .full() .... 一样工作。但是这个实现中存在错误。定义一个简单的“while not q.empty(): q.get()”循环通常会显示队列为空，即使您已经将数千个对象推入其中。我应该报告这些（但不会保证会得到它，所以如果你愿意，请随意）。
是的，当我说.full 不可靠时，这就是我的意思：在Queue 识别它是空的还是满的之前似乎存在延迟。它确实会引发明显的错误（mutiprocessing.queues.Full 和multiprocessing.queues.Empty），但至少Empty 异常似乎与.empty() 方法存在相同的问题。不过，到目前为止，在我的测试中，调用q.put(i,block=False) 似乎可以可靠地通过mutiprocessing.queues.Full 异常，因此包装类实际上可以工作...

【解决方案3】：

在 MacOSX 上为我工作

>>> import Queue
>>> Queue.Queue(30000000)
<Queue.Queue instance at 0x1006035f0>

【讨论】：

感谢 Sentinel -- 我使用的是多处理队列。也许我应该停止这样做？
好吧，如果您在应用程序中使用多处理，您会希望使用正确的类，而不仅仅是规避问题...
@Jason：这两个队列可能看起来相同，但实际上并非如此。 Multiprocessing 的 Queue 是使用 Pipe 实现的，而常规 Queue 本质上是一个带锁的出队。您实际上确实需要前者来执行 IPC，多处理就是这种情况。除非您自己重新实现，否则您可能会遇到大小限制。
@kaloyan，谢谢。正如您所说，该应用程序完全不能使用常规队列。
@JasonSundram：不要将它们视为“常规队列”。它们是线程队列（对于我们来说，使用标准库中的 Python 线程模块。它们可能应该被组织为线程下的子模块（正如稍后使用 multiprocessing.Queue 所做的那样）。但这是历史上的产物之一。