python中的多处理加速函数

Question

我对Python多处理感到困惑。在

我试图加速一个从数据库中处理字符串的函数，但我一定误解了多处理是如何工作的，因为这个函数在给一个工人池时比用“正常处理”要花更长的时间。在

这里有一个我正在努力实现的例子。在

from time import clock, time
from multiprocessing import Pool, freeze_support

from random import choice


def foo(x):
    TupWerteMany = []
    for i in range(0,len(x)):
         TupWerte = []
          s = list(x[i][3])
          NewValue = choice(s)+choice(s)+choice(s)+choice(s)
          TupWerte.append(NewValue)
          TupWerte = tuple(TupWerte)

          TupWerteMany.append(TupWerte)
     return TupWerteMany



 if __name__ == '__main__':
     start_time = time()
     List = [(u'1', u'aa', u'Jacob', u'Emily'),
        (u'2', u'bb', u'Ethan', u'Kayla')]
     List1 = List*1000000

     # METHOD 1 : NORMAL (takes 20 seconds) 
     x2 = foo(List1)
     print x2[1:3]

     # METHOD 2 : APPLY_ASYNC (takes 28 seconds)
     #    pool = Pool(4)
     #    Werte = pool.apply_async(foo, args=(List1,))
     #    x2 = Werte.get()
     #    print '--------'
     #    print x2[1:3]
     #    print '--------'

     # METHOD 3: MAP (!! DOES NOT WORK !!)

     #    pool = Pool(4)
     #    Werte = pool.map(foo, args=(List1,))
     #    x2 = Werte.get()
     #    print '--------'
     #    print x2[1:3]
     #    print '--------'


     print 'Time Elaspse: ', time() - start_time

我的问题：

1. 为什么apply_async比“正常方式”耗时更长？在
2. 我怎么把地图弄错了？在
3. 用多处理来加速这些任务有意义吗？在
4. 最后：在我读了这么多之后，我想知道python中的多处理在windows上是否有效？在

Answer 1

如果您感兴趣，这里有一个通用的多处理模板。在

import multiprocessing as mp
import time

def worker(x):
    time.sleep(0.2)
    print "x= %s, x squared = %s" % (x, x*x)
    return x*x

def apply_async():
    pool = mp.Pool()
    for i in range(100):
        pool.apply_async(worker, args = (i, ))
    pool.close()
    pool.join()

if __name__ == '__main__':
    apply_async()

输出如下：

^{pr2}$

如您所见，这些数字不是按顺序排列的，因为它们是异步执行的。在

Answer 2

因此，您的第一个问题是foo(x)中没有实际的并行，您只需将整个列表传递给函数一次。在

（一） 进程池的思想是让许多进程在某些数据的独立位上进行计算。在

 # METHOD 2 : APPLY_ASYNC
 jobs = 4
 size = len(List1)
 pool = Pool(4)
 results = []
 # split the list into 4 equally sized chunks and submit those to the pool
 heads = range(size/jobs, size, size/jobs) + [size]
 tails = range(0,size,size/jobs)
 for tail,head in zip(tails, heads):
      werte = pool.apply_async(foo, args=(List1[tail:head],))
      results.append(werte)

 pool.close()
 pool.join() # wait for the pool to be done

 for result in results:
      werte = result.get() # get the return value from the sub jobs

这只会给你一个实际的加速，如果它需要处理每个块的时间大于它启动进程的时间，在四个进程和四个作业要做的情况下，当然，如果你有4个进程和100个作业要做，这些动态会改变。请记住，您要创建一个全新的python解释器四次，这不是免费的。在

2）map的问题是它在一个单独的过程中将foo应用到{}中的每个元素，这将花费相当长的时间。因此，如果你有4个进程，map将弹出列表中的一个项目四次并将其发送给一个要处理的进程-等待进程完成-弹出列表中的其他内容-等待进程完成。这只有在处理单个项目需要很长时间时才有意义，例如，如果每个项目都是指向1GB文本文件的文件名。但是，map只接受列表的一个字符串并将其传递给foo，其中asapply_async获取列表的一部分。请尝试以下代码

^{pr2}$

这是内置的python映射，它将运行单个进程，但对于多进程版本，其思想是完全相同的。在

根据J.F.Sebastian的注释添加：但是您可以使用chunksize参数来为map指定每个块的近似大小。在

pool.map(foo, List1, chunksize=size/jobs) 

我不知道Windows上的map是否有问题，因为我没有一个可以测试的。在

3）是的，考虑到你的问题已经足够大了，有理由开发新的python解释器

4）不能给你一个明确的答案，因为这取决于核心/处理器的数量等，但一般来说，在Windows上应该没问题。在

Answer 3

关于问题（2） 在Dougal和Matti的指导下，我发现了问题所在。 最初的foo函数处理一个列表列表，而map需要一个函数来处理单个元素。在

新功能应该是

def foo2 (x):
    TupWerte = []
    s = list(x[3])
    NewValue = choice(s)+choice(s)+choice(s)+choice(s)
    TupWerte.append(NewValue)
    TupWerte = tuple(TupWerte)
    return TupWerte

还有叫它的街区：

^{pr2}$

感谢所有帮助我理解这一点的人。在

所有方法的结果： 对于List*2mio记录：普通13.3秒，与async并行：7.5秒，使用chuncksize与map并行：7.3，不使用chunksize 5.2秒