使用多处理从队列中读取

Question

以下是我使用Python多处理填充队列和从队列读取的代码：

from multiprocessing import Lock, Process, Queue, Pool
import time
from random import randint
def add_to_queue(tasks_to_accomplish, name):
while True:
random_int = randint(0, 22)
print('name', name , "adding" , random_int)
tasks_to_accomplish.put(random_int)
time.sleep(2)
def read_from_queue(tasks_to_accomplish, name):
while True:
item = tasks_to_accomplish.get()
print('name' , name , item)
time.sleep(.01)

if __name__ == '__main__':
tasks_to_accomplish = Queue()
p = Process(target=add_to_queue, args=(tasks_to_accomplish, "p"))
p.start()
p2 = Process(target=read_from_queue, args=(tasks_to_accomplish, "p2"))
p2.start()
p3 = Process(target=read_from_queue, args=(tasks_to_accomplish, "p3"))
p3.start()
p.join()
p2.join()
p3.join()

代码将无限执行，这里是部分输出：

name p adding 3
name p2 3
name p adding 4
name p3 4
name p adding 0
name p2 0
name p adding 22
name p3 22
name p adding 2
name p2 2
name p adding 13
name p3 13
name p adding 0
name p2 0
name p adding 14
name p3 14
name p adding 20
name p2 20
name p adding 4
name p3 4

从队列中读取所用时间为.01秒：time.sleep(.01)。但是p2&p3进程似乎不会在.01秒内读取线程，因为很明显它们阻塞的时间超过了.01秒。我是否正确地实现了从队列中读取的进程线程？

Answer 1

正如Daniel所指出的，Queue.get((将阻塞，直到默认情况下数据可用。

您可以使用q.get(block=True)来更改，尽管这会引发一个异常：

name p adding 12
name p2 12
Process Process-6:
Traceback (most recent call last):
File "/home/user/.pyenv/versions/3.8.0/lib/python3.8/multiprocessing/process.py", line 313, in _bootstrap
self.run()
File "/home/user/.pyenv/versions/3.8.0/lib/python3.8/multiprocessing/process.py", line 108, in run
self._target(*self._args, **self._kwargs)
File "<ipython-input-2-4e6d57c64980>", line 15, in read_from_queue
item = tasks_to_accomplish.get(block=False)
File "/home/user/.pyenv/versions/3.8.0/lib/python3.8/multiprocessing/queues.py", line 110, in get
raise Empty
_queue.Empty
Process Process-5:
Traceback (most recent call last):
File "/home/user/.pyenv/versions/3.8.0/lib/python3.8/multiprocessing/process.py", line 313, in _bootstrap
self.run()
File "/home/user/.pyenv/versions/3.8.0/lib/python3.8/multiprocessing/process.py", line 108, in run
self._target(*self._args, **self._kwargs)
File "<ipython-input-2-4e6d57c64980>", line 15, in read_from_queue
item = tasks_to_accomplish.get(block=False)
File "/home/user/.pyenv/versions/3.8.0/lib/python3.8/multiprocessing/queues.py", line 110, in get
raise Empty
_queue.Empty
name p adding 2
name p adding 12
name p adding 14
name p adding 21
name p adding 9
name p adding 13

您需要：

def read_from_queue(tasks_to_accomplish, name):
while True:
try:
item = tasks_to_accomplish.get(block=False)
except:
print('no data for', name)
else:
print('name' , name , item)

time.sleep(.01)

获取：

name p adding 0
name p2 0
no data for p3
no data for p3
no data for p2
no data for p2
no data for p3
no data for p2
no data for p3
# about 350 more entries like this
name p adding 5
no data for p2
name p3 5
no data for p2
no data for p3
no data for p3
no data for p2
no data for p3
# ...

除非您需要在读取之间做一些工作，否则我会说是，您已经正确地实现了读取过程(并且在读取时可以安全地删除对sleep的调用(。