Python多进程队列中的死锁问题

Question

我正在使用multiprocessing库中的队列在进程间共享数据。

我有两个队列，每个队列限制为10个对象。第一个队列有一个进程将对象“放入”其中，许多进程从中“获取”。第二个队列有许多进程将对象“放入”其中，只有一个进程从中“获取”。

系统一段时间内运行得非常完美，然后开始表现出奇怪的行为：“放置”对象到第一个队列中的进程继续工作，而从第一个队列读取的进程似乎不再工作了（尽管这些进程仍然存活）。这里似乎发生了死锁，但我不确定，以下是我的代码：

import multiprocessing
import logging
from multiprocessing import Process

logger = logging.get_logger(__name__)

# Processes 2, 3 ,4:

class Processes_234(Process):
    def __init__(self, message_queue_1, message_queue_2):
        Process.__init__(self)
        self.message_queue_1 = message_queue_1
        self.message_queue_2 = message_queue_2

    def run(self):
        while True:
            try:
                # get from queue
                el1, el2, el3 = self.message_queue_1.get()
                logger.debug('Processes234: get from queue')
            except Exception as exp:
                logger.debug("message_queue_1: queue empty, Exception message: " + str(exp))

            # do some stuff with el1, el2, el3...

            try:
                # put into second queue
                self.message_queue_2.put_nowait((el1, el2, el3))
                logger.debug('Processes234: put into queue')
            except Exception as excpt:
                logger.debug(excpt)
                logger.debug("message_queue_2: queue is full")
                # the queue is full so replace the old element with the new one
                try:
                    self.message_queue_2.get_nowait()
                    self.message_queue_2.put_nowait((el1, el2, el3))
                    # in case other process already fill the queue - ignore
                except:
                    pass


# process 5:
class Process5(Process):
    def __init__(self, message_queue_2):
        Process.__init__(self)
        self.message_queue_2 = message_queue_2

    def run(self):
        while True:
            try:
                # get from queue
                el1, el2, el = self.message_queue_2.get()
                print('Process5: get from queue')

            except Exception as exp:
                print("message_queue_2: queue empty, Exception message: " + str(exp))


def start_process_1():
    # init queues
    message_queue_1 = multiprocessing.Queue(maxsize=10)
    message_queue_2 = multiprocessing.Queue(maxsize=10)

    processes_234 = [Processes_234(message_queue_1, message_queue_2)
                     for _ in range(3)]

    for proc in processes_234:
        proc.start()

    process5 = Process5(message_queue_2)
    process5.start()
    counter = 1

    while True:
        el1 = counter + 1
        el2 = counter + counter
        el3 = "some string " * ((counter ** 2) % 60000)
        counter += 1
        # start passing data
        try:

            # put into queue
            message_queue_1.put_nowait((el1, el2, el3))
            logger.debug('Process1: put into queue')
        except Exception as excpt:
            logger.debug(excpt)
            logger.debug("message_queue_1: queue is full")
            # the queue is full so replace the old element with the new one
            try:
                message_queue_1.get_nowait()
                message_queue_1.put_nowait((el1, el2, el3))
                # in case other process already fill the queue - ignore
            except:
                pass


if __name__ == '__main__':
    start_process_1()

有人知道我的问题吗？

我正在使用Python 3.6.5。

- Yinon_90

---

我看不出你发布的内容有任何明显问题，但是很难重现你的问题，因为我们必须假设代码的执行方式（你的Process*对象和__main__是什么样子）。请创建一个MCVE，并解释一下你如何确定你的Process_234对象停止工作。 - shmee

谢谢，我解决了，并且根据您的要求更新了代码以重现此情况。 - Yinon_90

Answer 1

最终我解决了问题，原来是日志记录器！根据logging库的说法，日志记录器是线程安全的，但不是多进程安全的。

我修改了代码，使每个进程都有自己的日志记录器，并解决了这个问题。

- Yinon_90

---

很好的发现，感谢您更新了您的问题。这样做可能会帮助其他遇到类似问题的人。 - shmee

如果我没有找到这个，我不确定要花多长时间才能解决这个问题，谢谢！ - Mazyod