多处理队列 - 子进程有时会卡住并且不会收获

Question

首先，如果标题有点奇怪，我深表歉意，但我真的想不出如何将我面临的问题放在一行中。

所以我有以下代码

import time
from multiprocessing import Process, current_process, Manager
from multiprocessing import JoinableQueue as Queue

# from threading import Thread, current_thread
# from queue import Queue


def checker(q):
    count = 0
    while True:
        if not q.empty():
            data = q.get()
            # print(f'{data} fetched by {current_process().name}')
            # print(f'{data} fetched by {current_thread().name}')
            q.task_done()
            count += 1
        else:
            print('Queue is empty now')
            print(current_process().name, '-----', count)
            # print(current_thread().name, '-----', count)


if __name__ == '__main__':
    t = time.time()
    # m = Manager()
    q = Queue()
    # with open("/tmp/c.txt") as ifile:
    #     for line in ifile:
    #         q.put((line.strip()))
    for i in range(1000):
        q.put(i)
    time.sleep(0.1)
    procs = []
    for _ in range(2):
        p = Process(target=checker, args=(q,), daemon=True)
        # p = Thread(target=checker, args=(q,))
        p.start()
        procs.append(p)
    q.join()
    for p in procs:
        p.join()

示例输出

1：当进程刚刚挂起时

Queue is empty now
Process-2 ----- 501
output hangs at this point

2：当一切正常时。

Queue is empty now
Process-1 ----- 515
Queue is empty now
Process-2 ----- 485

Process finished with exit code 0

现在这种行为是间歇性的，有时会发生，但并非总是如此。

我也尝试使用Manager.Queue() 来代替multiprocessing.Queue()，但没有成功，并且都出现了同样的问题。

我用multiprocessing 和multithreading 对此进行了测试，我得到了完全相同的行为，与multithreading 相比，与multiprocessing 相比，这种行为的发生率要低得多。

所以我认为我在概念上遗漏了一些东西或做错了，但我现在无法抓住它，因为我在这方面花费了太多时间，现在我的脑海中没有看到可能非常基本的东西。

感谢您的帮助。

Answer 1

我相信您在 checker 方法中存在竞争条件。您检查队列是否为空，然后在单独 步骤中将下一个任务出列。将这两种操作分开而不互斥或锁定通常不是一个好主意，因为队列的状态可能会在检查和弹出之间发生变化。它可能是非空的，但另一个进程可能会在通过检查的进程能够这样做之前将等待的工作出列。

但是，我通常更喜欢沟通而不是锁定；它更不容易出错，并使一个人的意图更清晰。在这种情况下，我会向工作进程发送一个标记值（例如None），以指示所有工作都已完成。然后每个工作人员将下一个对象（始终是线程安全的）出列，如果对象是None，则子进程退出。

下面的示例代码是您的程序的简化版本，应该可以在没有竞争的情况下工作：

def checker(q):
    while True:
        data = q.get()
        if data is None:
            print(f'process f{current_process().name} ending')
            return
        else:
            pass # do work

if __name__ == '__main__':
    q = Queue()
    for i in range(1000):
        q.put(i)
    procs = []
    for _ in range(2):
        q.put(None) # Sentinel value
        p = Process(target=checker, args=(q,), daemon=True)
        p.start()
        procs.append(p)
    for proc in procs:
        proc.join()