Python多生产者单消费者的多进程任务处理

在工作中会遇到多生产者单消费者的任务，任务模型如下：

任务模型

Python实现如下：

from multiprocessing import Process, JoinableQueue
import random
import time


def producer_task(p_list, q):
    '''
    多进程处理任务，并将请求结果put到消息队列中
    :param p_list: 单个进程需要处理的任务
    :param q: 消息队列
    '''
    for p in p_list:
        time.sleep(random.randint(1, 3))
        q.put({'message': p})
    q.join()


def consumer_task(q):
    '''
    单进程处理消息队列中的信息
    :param q: 消息队列
    '''
    while True:
        message = q.get()
        print(message)
        q.task_done()


if __name__ == '__main__':

    p1_list = ['p1_1', 'p1_2', 'p1_3', 'p1_4', 'p1_5']
    p2_list = ['p2_1', 'p2_2', 'p2_3', 'p2_4', 'p2_5']
    p3_list = ['p3_1', 'p3_2', 'p3_3', 'p3_4', 'p3_5']

    q = JoinableQueue()
    p1 = Process(target=producer_task, args=(p1_list, q))
    p2 = Process(target=producer_task, args=(p2_list, q))
    p3 = Process(target=producer_task, args=(p3_list, q))

    # 保存到数据库的进程只能有一个
    c = Process(target=consumer_task, args=(q,))
    c.daemon = True  # c 设置为守护进程

    # 启动进程
    p_c = [p1, p2, p3, c]
    for p in p_c:
        p.start()

    p1.join()
    p2.join()
    p3.join()