使用while循环进行基本多处理

Question

我是multiprocessingpython中的新软件包,对于那些了解更多内容的人来说,我的困惑可能很容易.我一直在阅读有关并发的内容,并且已经搜索了其他类似的问题并且一无所获.(仅供参考我不希望使用multithreading,因为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()

此代码生成以下内容:

even - do nothing

任何从明智到这个问题的洞察力,我的代码或推理都不尽如人意,将不胜感激.

Answer 1

这是一种我曾多次使用并获得成功的方法:

1. 启动多处理池.

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使用相同的目标函数和队列进行多次调用,以使多个工作程序在同一队列上工作.只要确保你给队列足够的毒丸,以便所有的工人得到一个.

限制

您希望在队列中传递的任何数据必须由pickle模块进行选择(可序列化).看这里看看哪些可以和不可以腌制.

可能还有其他一些限制,但我无法想到任何其他限制因素.