我一直在使用multiprocessing包，并注意到在以下情况下，队列可能会出现死锁读取：
The "reader" process is using get with timeout > 0:
self.queue.get(timeout=3)"reader" dies while get is blocking due to timeout.
之后队列将永久锁定。
演示该问题的应用程序
我创建了两个子进程 "Worker"（将任务放入队列）和 "Receiver"（从队列获取任务）。此外，父进程定期检查他的子进程是否仍在运行，必要时启动新的子进程。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import multiprocessing
import procname
import time
class Receiver(multiprocessing.Process):
''' Reads from queue with 3 secs timeout '''
def __init__(self, queue):
multiprocessing.Process.__init__(self)
self.queue = queue
def run(self):
procname.setprocname('Receiver')
while True:
try:
msg = self.queue.get(timeout=3)
print '<<< `{}`, queue rlock: {}'.format(
msg, self.queue._rlock)
except multiprocessing.queues.Empty:
print '<<< EMPTY, Queue rlock: {}'.format(
self.queue._rlock)
pass
class Worker(multiprocessing.Process):
''' Puts into queue with 1 sec sleep '''
def __init__(self, queue):
multiprocessing.Process.__init__(self)
self.queue = queue
def run(self):
procname.setprocname('Worker')
while True:
time.sleep(1)
print 'Worker: putting msg, Queue size: ~{}'.format(
self.queue.qsize())
self.queue.put('msg from Worker')
if __name__ == '__main__':
queue = multiprocessing.Queue()
worker = Worker(queue)
worker.start()
receiver = Receiver(queue)
receiver.start()
while True:
time.sleep(1)
if not worker.is_alive():
print 'Restarting worker'
worker = Worker(queue)
worker.start()
if not receiver.is_alive():
print 'Restarting receiver'
receiver = Receiver(queue)
receiver.start()
在ps中，进程树是什么样子的
bash
\_ python queuetest.py
\_ Worker
\_ Receiver
控制台输出
$ python queuetest.py
Worker: putting msg, Queue size: ~0
<<< `msg from Worker`, queue rlock: <Lock(owner=None)>
Worker: putting msg, Queue size: ~0
<<< `msg from Worker`, queue rlock: <Lock(owner=None)>
Restarting receiver <-- killed Receiver with SIGTERM
Worker: putting msg, Queue size: ~0
Worker: putting msg, Queue size: ~1
Worker: putting msg, Queue size: ~2
<<< EMPTY, Queue rlock: <Lock(owner=SomeOtherProcess)>
Worker: putting msg, Queue size: ~3
Worker: putting msg, Queue size: ~4
Worker: putting msg, Queue size: ~5
<<< EMPTY, Queue rlock: <Lock(owner=SomeOtherProcess)>
Worker: putting msg, Queue size: ~6
Worker: putting msg, Queue size: ~7
有没有什么方法可以绕过这个问题？使用get_nowait和sleep的组合似乎是一种解决方法，但它不能“随时读取”数据。
系统信息
$ uname -sr
Linux 3.11.8-200.fc19.x86_64
$ python -V
Python 2.7.5
In [3]: multiprocessing.__version__
Out[3]: '0.70a1'
“一切都能正常工作”的解决方案
在撰写这篇问题时，我对Receiver类做出了一些愚蠢的修改：
class Receiver(multiprocessing.Process):
def __init__(self, queue):
multiprocessing.Process.__init__(self)
self.queue = queue
def run(self):
procname.setprocname('Receiver')
while True:
time.sleep(1)
while True:
try:
msg = self.queue.get_nowait()
print '<<< `{}`, queue rlock: {}'.format(
msg, self.queue._rlock)
except multiprocessing.queues.Empty:
print '<<< EMPTY, Queue rlock: {}'.format(
self.queue._rlock)
break
但是我觉得这并不是很好。