Python 多进程模块 multiprocessing - 简单使用

示例代码

问题：随机生成 $10^4$ 个在$[0,100)$之间的整数，获取每个数的平均值，并把所有结果写入文件中。

import numpy as np
import multiprocessing


def power(numbers, list_obj):
    """
    :param numbers: 输入的单个chunk的随机数字
    :param list_obj: 用于保存最终结果的列表
    :return:
    """
    res = []
    for number in numbers:
        res.append(number**2)
    list_obj.append(res)


if __name__ == '__main__':
    # 指定进程数量
    n_cpu = 20
    # 创建一个服务进程，用于多进程间的通信（保存每个进程的结果），这里创建list对象来保存处理的结果。
    mgr = multiprocessing.Manager()
    list_obj = mgr.list()
    # 随机生成数据并将数据分为 n_cpu 份
    numbers = np.random.randint(low=0, high=100, size=int(10e4))
    chunks = [list(i) for i in np.array_split(numbers, n_cpu)]
    # 创建进程实例，并绑定其函数以及需要传入的数据
    jobs = []
    for i in range(n_cpu):
        p = multiprocessing.Process(target=power,
                                    args=(chunks[i], list_obj))
        jobs.append(p)
        # 开始
        p.start()

    # 等待所有进程完成
    for p in jobs:
        p.join()

    # 整合处理后的数据
    res = []
    for i in list_obj:
        res += i
    res = np.array(res).astype(int)

    # 保存处理后的结果
    np.savetxt('res.txt', res, fmt='%d')
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

1.相关知识

服务进程

由 Manager() 返回的管理器对象控制一个服务器进程，该进程保存Python对象并允许其他进程使用代理操作它们。
Manager() 返回的管理器支持类型： list 、 dict 、 Namespace 、 Lock 、 RLock 、 Semaphore 、 BoundedSemaphore 、 Condition 、 Event 、 Barrier 、 Queue 、 Value 和 Array。

2.参考

numpy.random.randint https://numpy.org/doc/stable/reference/random/generated/numpy.random.randint.htm
numpy.array_split
https://numpy.org/doc/stable/reference/generated/numpy.array_split.html
numpy.savetxt
https://numpy.org/doc/stable/reference/generated/numpy.savetxt.html
multiprocessing
https://docs.python.org/3/library/multiprocessing.html