Python：多核处理？答案

【问题标题】：Python: Multicore processing?Python：多核处理？
【发布时间】：2010-11-14 00:23:34
【问题描述】：

我一直在阅读 Python 的 multiprocessing module。我仍然不认为我对它可以做什么有很好的了解。

假设我有一个四核处理器，我有一个包含 1,000,000 个整数的列表，我想要所有整数的总和。我可以这样做：

list_sum = sum(my_list)

但这只会将它发送到一个核心。

是否有可能，使用多处理模块，将数组划分，让每个核心得到它的部分的总和并返回值，以便计算总和？

类似：

core1_sum = sum(my_list[0:500000])          #goes to core 1
core2_sum = sum(my_list[500001:1000000])    #goes to core 2
all_core_sum = core1_sum + core2_sum        #core 3 does final computation

任何帮助将不胜感激。

【问题讨论】：

标签： python multicore multiprocessing

【解决方案1】：

是的，可以对多个进程进行求和，就像使用多个线程一样：

from multiprocessing import Process, Queue

def do_sum(q,l):
    q.put(sum(l))

def main():
    my_list = range(1000000)

    q = Queue()

    p1 = Process(target=do_sum, args=(q,my_list[:500000]))
    p2 = Process(target=do_sum, args=(q,my_list[500000:]))
    p1.start()
    p2.start()
    r1 = q.get()
    r2 = q.get()
    print r1+r2

if __name__=='__main__':
    main()

但是，使用多个进程执行此操作可能比使用单个进程执行此操作要慢，因为来回复制数据比立即求和更昂贵。

【讨论】：

@Martin，我相信这个死锁，每个docs.python.org/library/…：“一个将项目放入队列的进程将在终止之前等待，直到所有缓冲的项目都由“馈送”线程馈送到底层管道”——文档给出的死锁示例与您的代码非常相似（它是开始、连接、获取序列中的单个子进程）和两个子进程而不是一个没有帮助。交换连接和获取，或者只是删除连接。
“它对我有用”，可能是因为数据只是适合管道。无论如何，我已经删除了连接。
你是在 Linux 上运行这个吗？
是的，我确实在 Linux 上运行过。
除非你把 if name == "main"

【解决方案2】：

欢迎来到并发编程的世界。

Python 能做什么（不能做什么）取决于两件事。

操作系统可以（和不能）做什么。大多数操作系统将进程分配给内核。要使用 4 个核心，您需要将问题分解为四个进程。这比听起来容易。有时。
底层 C 库可以（和不能）做什么。如果 C 库公开了 OS 的功能并且 OS 公开了硬件的功能，那么你就是可靠的。

将一个问题分解为多个进程——尤其是在 GNU/Linux 中——很容易。将其分解为多步骤管道。

在求和一百万个数字的情况下，想想下面的 shell 脚本。假设某个假设的 sum.py 程序对标准输入上的数字范围或数字列表求和。

(sum.py 0 500000 & sum.py 50000 1000000) | sum.py

这将有 3 个并发进程。两个是对很多数字求和，第三个是对两个数字求和。

由于 GNU/Linux shell 和操作系统已经为您处理了部分并发，您可以设计简单（非常、非常简单）的程序，从标准输入读取、写入标准输出，并设计用于执行一小部分大工作。

您可以尝试通过使用subprocess 构建管道而不是将作业分配给shell 来减少开销。但是，您可能会发现 shell 非常非常快速地构建管道。（它是直接用 C 语言编写的，可以为您直接调用 OS API。）

【讨论】：

我觉得这个答案显示了很多独创性。 CS中没有什么问题是简单的加一层间接层解决不了的。
@earino：OTOH，它根本没有回答 OP 的问题，具体是关于“我如何使用多处理模块”。
@Martin v. Löwis：是的。 IMO 更大的问题（使用所有内核）通常比提出的问题（使用子进程来使用所有内核）更重要。在某些情况下，提出的问题反映了一些糟糕的假设。

【解决方案3】：

当然，例如：

from multiprocessing import Process, Queue

thelist = range(1000*1000)

def f(q, sublist):
    q.put(sum(sublist))

def main():
    start = 0
    chunk = 500*1000
    queue = Queue()
    NP = 0
    subprocesses = []
    while start < len(thelist):
      p = Process(target=f, args=(queue, thelist[start:start+chunk]))
      NP += 1
      print 'delegated %s:%s to subprocess %s' % (start, start+chunk, NP)
      p.start()
      start += chunk
      subprocesses.append(p)
    total = 0
    for i in range(NP):
      total += queue.get()
    print "total is", total, '=', sum(thelist)
    while subprocesses:
      subprocesses.pop().join()

if __name__ == '__main__':
    main()

结果：

$ python2.6 mup.py 
delegated 0:500000 to subprocess 1
delegated 500000:1000000 to subprocess 2
total is 499999500000 = 499999500000

请注意，这种粒度太细，不值得生成进程——整体求和任务很小（这就是为什么我可以重新计算 main 中的总和作为检查；-）并且太多数据被移回并且第四（事实上，子进程不需要获取他们处理的子列表的副本——索引就足够了）。因此，这是一个“玩具示例”，其中并不需要真正保证多处理。然而，使用不同的架构（使用从队列接收多个任务以执行的子进程池、最小化来回数据移动等）以及在粒度较小的任务上，您实际上可以获得性能方面的好处。

【讨论】：