python - 列表的并行处理

Question

我正在尝试了解多重处理。我有一个列表，我将其分为两个同样长的部分，我将它们分入两个单独的过程中进行排序。我知道这部分是有效的，因为打印 saveto 给了我两个列表。但我无法访问它们，因为最后我得到两个空列表。为什么我无法访问我想要写入 l1 和 l2 的内容以及如何操作？

import multiprocessing
import random


def sort(l, saveto):
    saveto = sorted(l)
    print saveto

if __name__ == '__main__':

    l = [int(100000*random.random()) for i in xrange(10000)]

    listlen = len(l)
    halflist = listlen/2
    l1 = []
    l2 = []

    p1 = multiprocessing.Process(target=sort, args=(l[0:halflist], l1))
    p2 = multiprocessing.Process(target=sort, args=(l[halflist:listlen], l2))

    p1.start()
    p2.start()

    p1.join()
    p2.join()

    print l1
    print l2

Answer 1

使用multiprocessing.Queue在进程之间共享数据

import multiprocessing
import random

def sort(l, queue):
    queue.put(sorted(l))


if __name__ == '__main__':
    l = [int(100000*random.random()) for i in xrange(10000)]

    listlen = len(l)
    halflist = listlen/2
    queue = multiprocessing.Queue()

    p1 = multiprocessing.Process(target=sort, args=(l[0:halflist], queue))
    p2 = multiprocessing.Process(target=sort, args=(l[halflist:listlen], queue))

    p1.start()
    p2.start()

    p1.join()
    p2.join()

    print queue.get()
    print queue.get()

更新:

事实证明，将大量数据放入队列可能会导致死锁。这是mentioned in the docs :

Warning

As mentioned above, if a child process has put items on a queue (and it has not used JoinableQueue.cancel_join_thread), then that process will not terminate until all buffered items have been flushed to the pipe.

This means that if you try joining that process you may get a deadlock unless you are sure that all items which have been put on the queue have been consumed. Similarly, if the child process is non-daemonic then the parent process may hang on exit when it tries to join all its non-daemonic children.

Note that a queue created using a manager does not have this issue.

修复版本:

import multiprocessing
import random

def sort(l, queue):
    queue.put(sorted(l))


if __name__ == '__main__':
    l = [int(100000*random.random()) for i in range(10000)]

    listlen = len(l)
    halflist = listlen/2

    manager = multiprocessing.Manager()
    queue = manager.Queue()

    p1 = multiprocessing.Process(target=sort, args=(l[0:halflist], queue))
    p2 = multiprocessing.Process(target=sort, args=(l[halflist:listlen], queue))

    p1.start()
    p2.start()

    p1.join()
    p2.join()

    print queue.get()
    print queue.get()