python - 使用 while 循环的基本多处理

Question

我是 python 中的 multiprocessing 包的新手，了解更多的人可能很容易理清我的困惑。我一直在阅读有关并发的信息，并搜索过类似的其他问题，但一无所获。 (仅供引用，我不想想使用多线程，因为 GIL 会大大降低我的应用程序的速度。)

我在事件的框架内思考。我想让多个进程运行，等待事件发生。如果事件发生，它会被分配给一个特定的进程，该进程运行然后返回到其空闲状态。可能有更好的方法来做到这一点，但我的理由是我应该生成所有进程一次并无限期地保持它们打开，而不是在每次事件发生时创建然后关闭进程。速度对我来说是个问题，我的事件每秒可能发生数千次。

我提出了以下玩具示例，旨在将偶数发送到一个进程，将奇数发送到另一个进程。这两个过程是相同的，它们只是将数字附加到列表中。

from multiprocessing import Process, Queue, Pipe

slist=['even','odd']

Q={}
Q['even'] = Queue()
Q['odd'] = Queue()

ev,od = [],[]

Q['even'].put(ev)
Q['odd'].put(od)

P={}
P['even'] = Pipe()
P['odd'] = Pipe()



def add_num(s):
    """ The worker function, invoked in a process. The results are placed in
        a list that's pushed to a queue."""
#    while True :
    if not P[s][1].recv():
        print s,'- do nothing'

    else:            
        d = Q[s].get()
        print d
        d.append(P[s][1].recv())
        Q[s].put(d)
        print Q[s].get()
        P[s][0].send(False)
        print 'ya'




def piper(s,n):

    P[s][0].send(n)    
    for k in [S for S in slist if S != s]:
        P[k][0].send(False) 
    add_num(s)


procs = [ Process (
                   target=add_num,
                   args=(i,)
                   ) 
         for i in ['even','odd']]

for s in slist: 
    P[s][0].send(False)

for p in procs:
    p.start()  
    p.join()

for i in range(10):
    print i
    if i%2==0:
        s = 'even'
    else:
        s = 'odd'
    piper(s,i)


print 'results:', Q['odd'].get(),Q['even'].get()

此代码产生以下内容:

甚至 - 什么都不做

如果我的代码或推理不足等问题，智者对这个问题的任何见解都将不胜感激。

Answer 1

这是我使用过几次并取得成功的方法:

1. 启动 multiprocessing pool .

2. 使用多处理 SyncManager创建多个队列(一个用于需要不同处理的每种数据类型)。

3. 使用 apply_async启动处理数据的功能。就像队列一样，对于需要以不同方式处理的每种类型的数据，应该有一个函数。每个启动的函数都将与其数据对应的队列作为输入参数。这些函数将在一个无限循环中完成它们的工作，该循环从从队列中获取数据开始。

4. 开始处理。在处理过程中，主进程对数据进行排序并决定哪个函数应该处理它。做出决定后，将数据放入该功能对应的队列中。

5. 处理完所有数据后，主进程将一个称为“毒丸”的值放入每个队列。毒丸是一个值，所有工作进程都将其识别为退出信号。由于队列是先进先出 (FIFO)，因此可以保证他们将毒丸作为队列中的最后一项拉出。

6. 关闭并加入多处理池。

代码

下面是这个算法的一个例子。示例代码的目标是使用前面描述的算法将奇数除以 2，将偶数除以 -2。所有结果都放在主进程可以访问的共享列表中。

import multiprocessing

POISON_PILL = "STOP"

def process_odds(in_queue, shared_list):

    while True:

        # block until something is placed on the queue
        new_value = in_queue.get() 

        # check to see if we just got the poison pill
        if new_value == POISON_PILL:
            break

        # we didn't, so do the processing and put the result in the
        # shared data structure
        shared_list.append(new_value/2)

    return

def process_evens(in_queue, shared_list):

    while True:    
        new_value = in_queue.get() 
        if new_value == POISON_PILL:
            break

        shared_list.append(new_value/-2)

    return

def main():

    # create a manager - it lets us share native Python object types like
    # lists and dictionaries without worrying about synchronization - 
    # the manager will take care of it
    manager = multiprocessing.Manager()

    # now using the manager, create our shared data structures
    odd_queue = manager.Queue()
    even_queue = manager.Queue()
    shared_list = manager.list()

    # lastly, create our pool of workers - this spawns the processes, 
    # but they don't start actually doing anything yet
    pool = multiprocessing.Pool()

    # now we'll assign two functions to the pool for them to run - 
    # one to handle even numbers, one to handle odd numbers
    odd_result = pool.apply_async(process_odds, (odd_queue, shared_list))
    even_result = pool.apply_async(process_evens, (even_queue, shared_list))
    # this code doesn't do anything with the odd_result and even_result
    # variables, but you have the flexibility to check exit codes
    # and other such things if you want - see docs for AsyncResult objects

    # now that the processes are running and waiting for their queues
    # to have something, lets give them some work to do by iterating
    # over our data, deciding who should process it, and putting it in
    # their queue
    for i in range(6):

        if (i % 2) == 0: # use mod operator to see if "i" is even
            even_queue.put(i)

        else:
            odd_queue.put(i)

    # now we've finished giving the processes their work, so send the 
    # poison pill to tell them to exit
    even_queue.put(POISON_PILL)
    odd_queue.put(POISON_PILL)

    # wait for them to exit
    pool.close()
    pool.join()

    # now we can check the results
    print(shared_list)

    # ...and exit!
    return


if __name__ == "__main__":
    main()

输出

此代码产生此输出:

[0.5, -0.0, 1.5, -1.0, 2.5, -2.0]

请注意，结果的顺序是不可预测的，因为我们无法保证函数能够以何种顺序从其队列中获取项目并将结果放入列表中。但您当然可以进行任何需要的后处理，其中可能包括排序。

理由

我认为这会很好地解决您的问题，因为:

1. 您说得对，生成进程的开销很大。这种单生产者/多消费者方法消除了在整个程序期间使用池来保持工作人员存活的情况。

2. 它解决了您对能够根据数据的属性以不同方式处理数据的担忧。在您的评论中，您表达了对能够将数据发送到特定进程的担忧。在这种方法中，您可以选择将数据提供给哪些进程，因为您必须选择将数据放在哪个队列中。 (顺便说一下，我认为您正在考虑 pool.map 函数，正如您正确地相信的那样，它不允许您在同一作业中执行不同的操作。apply_async 允许。)

3. 我发现它非常可扩展和灵活。需要添加更多类型的数据处理？只需编写您的处理函数，再添加一个队列，并将逻辑添加到 main 以将数据路由到您的新函数。您是否发现一个队列正在备份并成为瓶颈？您可以使用相同的目标函数调用 apply_async 并多次排队，让多个工作人员在同一个队列上工作。只要确保你给队列足够的毒丸，这样所有的 worker 都能得到一颗。

限制

您想要在队列中传递的任何数据都必须由 pickle 模块进行 picklable(可序列化)。看here看看什么可以腌制，什么不能腌制。

可能还有其他限制，但我想不出任何其他限制。