如何实现具有多个消费者的单个队列

Question

我有一个列表list_of_params 和一个函数run_train()，它接收来自list_of_params 的项目（例如run_train(list_of_params[0])）。我可以一次将run_train() 发送到多个 GPU。所以我想知道是否有任何可以并行化的单个队列的实现。

如果这还不够清楚，请想象以下场景：

“超市有一个顾客队列，但有5个收银员。一旦一个收银员空闲，它就会处理队列中下一个顾客的产品。这与每个收银员都有自己的线路相反。”

如果需要，我可以提供更多详细信息。

谢谢！

Answer 1

试试queue 模块。 '线程安全的多生产者、多消费者队列'。

Answer 2

下面是一个使用带有几个生产者和消费者进程的队列的示例：

from multiprocessing import Process, Queue, Event
#if you use threading instead of multiprocessing, use queue.Queue rather than multiprocessing.Queue
from queue import Empty
from time import sleep
from random import random

def producer(stopflag, out_queue):
    while True:
        if stopflag.is_set():
            break
        sleep(random()) #sleep anywhere from 0-1 sec avg 0.5 (producer will produce on average a maximum of 2 outputs / sec)
        out_queue.put(random()) #may wait if the queue is currently full, thereby reducing production rate.
        
def consumer(stopflag, in_queue):
    while True:
        if stopflag.is_set():
            break
        x = in_queue.get()
        sleep(x*2) #consumers work twice slower than producers averaging 1 item per sec

def main():
    stopflag = Event()
    
    shared_q = Queue(maxsize=100) # if the queue fills up, it will force the producers
                                  #   to wait for the consumers to catch up. Otherwise, it
                                  #   may grow infinitely (until the computer runs out of memory)
    
    #2 producers
    producers = [Process(target=producer, args=(stopflag, shared_q)) for _ in range(2)]
    #4 consumers
    consumers = [Process(target=consumer, args=(stopflag, shared_q)) for _ in range(4)]
    
    for process in producers + consumers:
        process.start()
        
    sleep(20) #let them work a while
    
    stopflag.set() #tell them to stop
    
    for process in producers + consumers: #wait for them to stop
        process.join()
        
    #empty unfinished tasks from the queue so its thread can exit normally
    #(it's good practice to clean up resources kind of like closing files when done with them)
    try:
        while True:
            shared_q.get_nowait()
    except Empty:
        pass
    
if __name__ == '__main__':
    main()