远程队列消费者在重启后错过了第一条消息

Question

我有如下代码：

server.py

import queue
from multiprocessing.managers import BaseManager

class QueueManager(BaseManager):
    pass

q = queue.Queue()
QueueManager.register('queue', callable=lambda:q)
m = QueueManager(address=('localhost', 51000), authkey=b'pass')
s = m.get_server()
s.serve_forever()

生产者.py

from multiprocessing.managers import BaseManager
import time

class QueueManager(BaseManager):
    pass

QueueManager.register('queue')
m = QueueManager(address=('localhost', 51000), authkey=b'pass')
m.connect()
queue = m.queue()

idx = 0
while True:
    time.sleep(2)
    queue.put(idx)
    idx += 1

consumer.py

from multiprocessing.managers import BaseManager

class QueueManager(BaseManager):
    pass

QueueManager.register('queue')
m = QueueManager(address=('localhost', 51000), authkey=b'pass')
m.connect()
queue = m.queue()

while True:
    message = queue.get()
    print(message)

如果我运行服务器和生产者，然后启动消费者，我会看到生产者放入队列的所有消息都出现在消费者身上。但是，如果我停止消费者并立即重新启动它，它总是会跳过一条消息。

为了说明我看到的 consumer.py 的输出：

0
1
2
3
<restart the consumer>
5
6
7
etc.

这正是 python 多处理队列应该如何工作的方式，是错误还是我做错了什么？

Answer 1

我认为问题在于管道在 python 中的实现方式，或者它甚至可能是对操作系统的限制。这是完整的堆栈跟踪：

Traceback (most recent call last):
  File "consumer.py", line 12, in <module>
    message = queue.get()
  File "<string>", line 2, in get
  File "/usr/local/Cellar/python3/3.6.0/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/managers.py", line 757, in _callmethod
    kind, result = conn.recv()
  File "/usr/local/Cellar/python3/3.6.0/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/connection.py", line 250, in recv
    buf = self._recv_bytes()
  File "/usr/local/Cellar/python3/3.6.0/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/connection.py", line 407, in _recv_bytes
    buf = self._recv(4)
  File "/usr/local/Cellar/python3/3.6.0/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/connection.py", line 379, in _recv
    chunk = read(handle, remaining)
KeyboardInterrupt

queue.get() 调用中的值似乎丢失了，该调用没有被SIGINT 正确终止。 queue.get() 立即被取消，因此 python 不会完成 get() 调用然后失去价值。看起来更像是 python 没有正确取消管道上的recv。

如果您将消费者更改为此：

while True:
    while queue.empty():
        sleep(0.1)
    message = queue.get()
    print(message)

它会起作用的。但当然这是一种解决方法，而不是真正的解决方案。

更新：

在更多地使用您的代码之后，我认为这是一个错误，因为：

1. 你followed their coding example one by one
2. 没有任何类型的队列可以解决问题（multiprocessing.Queue 和 multiprocessing.JoinableQueue
3. 同时发送 task_done() 也无济于事

该错误同时发生在 python2 和 python3 上。我建议你report this as a bug。在最坏的情况下，如果它不是一个错误，你至少可以解释为什么 python 会这样。