在python多处理池中实现生产者消费者模式的正确方法

如何解决在python多处理池中实现生产者消费者模式的正确方法

我想通过使用 producer-consumer pattern

来实现 multiprocessing.pool.Pool

由于 JoinableQueue 不能在 Pool 中使用（会声明为 RuntimeError: JoinableQueue objects should only be shared between processes through inheritance），所以我必须使用受 multiprocessing.Manager() 启发的 this answer。

问题是：现在当消费者工作比生产者工作多时程序可能会挂起。

import queue
import random
from multiprocessing import Manager,Pool


def consumer(q):
    while True:
        try:
            res = q.get(block=False)
            if res is None:
                break
            print(f'Consume {res}')
        except queue.Empty:
            pass


def producer(q,food):
    for i in range(2):
        res = f'{food} {i}'
        print(f'Produce {res}')
        q.put(res)
    q.put(None) # sentinel


if __name__ == "__main__":
    with Pool() as pool:
        jobs = 2
        foods = ['apple','banana','melon','salad']
        q = Manager().Queue()
        [pool.apply_async(func=consumer,args=(q,)) for _ in range(jobs + 1)]  # would hang

        # would not hang only when the consumer jobs is equal or less than the producer jobs
        # [pool.apply_async(func=consumer,)) for _ in range(jobs)] 

        [
            pool.apply_async(func=producer,random.choice(foods)))
            for _ in range(jobs)
        ]

        pool.close()
        pool.join()

似乎那些额外消费者无法获得哨兵​​m>而只能在那里永远等待。

那么在 multiprocessing.pool.Pool 中实现生产者-消费者模式的优雅方式是什么？

还是只能用multiprocessing.Process + JoinableQueue？

解决方法

您可以使用 multiprocessing.JoinableQueue，让您的进程池工作人员将其作为全局变量访问，该变量使用池初始值设定项进行初始化：

import multiprocessing

def init_pool(input_q,output_q):
    global in_q,out_q
    in_q = input_q
    out_q = output_q


def worker():
    print(type(in_q))


# required by Windows
if __name__ == '__main__':
    in_q = multiprocessing.JoinableQueue()
    out_q = multiprocessing.JoinableQueue()
    pool = multiprocessing.Pool(2,initializer=init_pool,initargs=(in_q,out_q))
    pool.apply(worker)

打印：

<class 'multiprocessing.queues.JoinableQueue'>

,

似乎使用 <script src="https://cdnjs.cloudflare.com/ajax/libs/jquery/3.3.1/jquery.min.js"></script> <div id="accordian"> <ul> <li class="active"> <a href="index.html">Augumented Reality Platform</a><span class="arrow">></span> <ul> <li><a href="#">Todo List</a></li> <li class="active"><a href="index.html">Namespaces</a> <span class="arrow">></span> <ul> <li class="active"><a href="#">Namespace List</a><span class="arrow">></span> <ul> <li><a href="#">anonymous_namespace</a></li> <li><a href="#">NCommon</a></li> <li><a href="#">std</a></li> <li><a href="#">units</a></li> </ul> </li> <li><a href="#">Namespace Members</a><span class="arrow">></span> <ul> <li><a href="#">All</a></li> <li><a href="#">Functions</a></li> <li><a href="#">Variables</a></li> <li><a href="#">Typedefs</a></li> <li><a href="#">Enumerations</a></li> </ul> </li> </ul> </li> <li><a href="#">Examples</a></li> </ul> </li> </ul> </div> + multiprocessing.Process 是一种更优雅的方式。

JoinableQueue