python多处理队列实现

Question

我无法理解如何在下面的多处理示例中实现队列。基本上，我希望代码：

1) 产生 2 个进程（完成）

2) 将我的 id_list 分成两部分（完成）

3) 让每个进程遍历列表并打印出每个项目，并且仅在完成列表后关闭。我知道我必须实现某种类型的排队系统，并将其传递给每个工人，但我不知道该怎么做。任何帮助将不胜感激。

from multiprocessing import Pool,Queue
id_list = [1,2,3,4,5,6,7,8,9,10]

def mp_worker(record):
    try:  
        print record
        sleep(1)
    except: pass
    print "worker closed"

def mp_handler():
    p = Pool(processes = 2) #number of processes
    p.map(mp_worker, id_list)  #devides id_list between 2 processes, defined above
    p.close()
    p.join()

mp_handler()

注意 - 代码会打印 10 次“工人已关闭”。我希望这条语句只打印两次（每个工人一次，在每个工人从 id_list 打印出 5 个数字之后）

Answer 1

使用 Pool 和 Queue 解决这个问题的一个方法是

    from time import sleep
    from multiprocessing import Pool,Queue
    id_list = [1,2,3,4,5,6,7,8,9,10]

    def mp_worker(q):
        try:  
            print(q.get())
            sleep(.1)
        except: pass
        print ("worker closed")

    if __name__ == "__main__":
        q = Queue()
        p = Pool(processes = 2) #number of processes
        for x in id_list:
            q.put(x)
        p.map(mp_worker, id_list)  #devides id_list between 2 processes, defined above

您必须在代码的主要部分中通过 put 将值添加到 Quene，并在函数中通过 get

从队列中读取值

Answer 2

你的打印语句误导了你——工作进程没有在函数结束时终止。事实上，工作进程在池关闭之前一直保持活动状态。此外，multiprocessing 已经负责将列表分解成块并为您安排每个任务。

至于你的其他问题，如果你想在整个 list 完成时触发异步事件，通常你会向map_async 传递一个回调。每个块调用一次需要对内部进行一些处理，但如果你真的想要你可以：

def mapstar_custom(args):
    result = list(map(*args))
    print "Task completed"
    return result
...

pool._map_async(f, x, mapstar_custom, None, None, None).get()

---

编辑：我们似乎在混淆术语。当我说工人时，我的意思是池产生的进程，而您似乎是指 Selenium 从这些进程中产生的进程（这不是您的问题）。只打开一次 webdriver 很容易：如果你有 pool.map(module.task, ...)，那么在 module.py 中就可以了：

# ... selenium init here ...

def task(...):
    # ... use webdriver ...

无论您分派该任务多少次，池工作人员都只会导入该模块一次。所以顶级初始化只会发生一次。

Answer 3

这对我有用（在 Python 3 上）。我没有使用池，而是生成了自己的两个进程：

from multiprocessing import Process, Queue
from time import sleep


id_list = [1,2,3,4,5,6,7,8,9,10]

queue = Queue()

def mp_worker(queue):

    while queue.qsize() >0 :
        record = queue.get()
        print(record)
        sleep(1)

    print("worker closed")

def mp_handler():

    # Spawn two processes, assigning the method to be executed 
    # and the input arguments (the queue)
    processes = [Process(target=mp_worker, args=(queue,)) for _ in range(2)]

    for process in processes:
        process.start()
        print('Process started')

    for process in processes:
        process.join()



if __name__ == '__main__':

    for id in id_list:
        queue.put(id)

    mp_handler()

虽然要处理的元素的长度是硬编码的。但它可能是 mp_worker 方法的第二个输入参数。