python - 池、队列、悬挂

Question

我想使用队列来保存结果，因为我希望消费者(串行而不是并行)在工作人员产生结果时处理工作人员的结果。

现在，我想知道为什么以下程序挂起。

import multiprocessing as mp
import time
import numpy as np
def worker(arg):
    time.sleep(0.2)
    q, arr = arg 
    q.put(arr[0])

p = mp.Pool(4)
x = np.array([4,4])
q = mp.Queue()

for i in range(4):
    x[0] = i 
    #worker((q,x))
    p.apply_async(worker, args=((q, x),)) 

print("done_apply")
time.sleep(0.2)
for i in range(4):
    print(q.get())

Answer 1

Queue对象不能共享。我首先找到了这个answer，得出了与OP相同的结论。 .

不幸的是，此代码中还存在其他问题(这不会使其与链接的答案完全相同)

• worker(arg)应该是worker(*arg)以便 args 解包工作。如果没有这个，我的进程也会被锁定(我承认我不知道为什么。它应该抛出异常，但我猜多处理和异常不能很好地协同工作)
• 通过相同的x对 worker 的结果与结果相同(对于 apply 有效，但对于 apply_async 无效

另一件事:为了使代码可移植，请将主代码包装在 if __name__ == "__main__": 中。 ，由于进程生成的差异，在 Windows 上是必需的

完全固定的代码，为我输出 0,3,2,1:

import multiprocessing as mp
import time
import numpy as np
def worker(*arg):  # there are 2 arguments to "worker"
#def worker(q, arr):  # is probably even better
    time.sleep(0.2)
    q, arr = arg
    q.put(arr[0])

if __name__ == "__main__":
    p = mp.Pool(4)

    m = mp.Manager()  # use a manager, Queue objects cannot be shared
    q = m.Queue()

    for i in range(4):
        x = np.array([4,4])  # create array each time (or make a copy)
        x[0] = i
        p.apply_async(worker, args=(q, x))

    print("done_apply")
    time.sleep(0.2)
    for i in range(4):
        print(q.get())