python多进程

当有多个非相关任务需要处理时，并行能大大提高处理速度。这里简要介绍python的multiprocessing模块。

---

简单多进程编写

当我们任务数量确定而且比较少的时候，可以手动为每个任务指定一个进程来运行。

import multiprocessing as mp

def f(a):
    print(a)

if __name__ == '__main__':
    # 这里有三个任务，手动指定3个进程
    p1 = mp.Process(target=f, args=(1,))
    p2 = mp.Process(target=f, args=(2,))
    p3 = mp.Process(target=f, args=(3,))
    p1.start()
    p2.start()
    p3.start()
    p1.join()
    p2.join()
    p3.join()

# 输出
1
2
3

使用进程池来处理多任务

当我们任务比较多而且不确定数量（又或者想使得代码更简洁）的时候可以使用进程池Pool来编写多进程。

import multiprocessing as mp

def f(a):
    return a

if __name__ == '__main__':
    pool = mp.Pool()
    res = pool.map(f, (1, 2, 3, 4))
    print(res)
# 输出
[1, 2, 3, 4]

可以看到，进程池不仅可以方便的处理多进程，还将各个进程的的处理结果储存了在一个列表中。Pool默认使用计算机所有cpu核来进行运算，也可使用Pool(process=4)来指定并行的进程数。
另一个可以储存结果的函数是apply_async()，但是这个函数只支持传入一个参数，也即运行一个任务，运行多个任务需要多次指定。此外，他返回的结果是一个类似生成器的东西，需要通过get函数取出来。

import multiprocessing as mp

def f(a):
    return a

if __name__ == '__main__':
    pool = mp.Pool()
    res = [pool.apply_async(f, (i,)) for i in range(4)]
    print([r.get() for r in res])

# 输出
[0, 1, 2, 3]

多参数函数的多进程

上面处理任务的函数都只有一个参数，在实际情况这种情况是很少的，一般我们的函数都需要传入多个参数。python2中未能支持传多个参数，python3.3后则有starmap来支持传入多参数。

import multiprocessing as mp

def f(a, b):
    return (a, b)

if __name__ == '__main__':
    pool = mp.Pool()
    res = pool.starmap(f, ((1, 2), ('a', 'b')))
    print(res)

# 输出
[(1, 2), ('a', 'b')]

另一个传多个参数的方法是通过python中的魔术方法*args:

import multiprocessing as mp

def f(*l):
    a, b = l[0]  # 为什么这里是l[0]而不是l?因为这里传入的l是一个嵌套元组（(a,b),）。
    return (a, b)

if __name__ == '__main__':
    pool = mp.Pool()
    res = pool.map(f, ((1, 2), ('a', 'b')))
    print(res)

# 输出
[(1, 2), ('a', 'b')]

亦或使用**kwargs:

import multiprocessing as mp

def f(*l):
    d = l[0]    # 这里l为（{'a': 1, 'b': 2},）
    return (d['a'], d['b'])

if __name__ == '__main__':
    pool = mp.Pool()
    res = pool.map(f, ({'a': 1, 'b': 2}, {'a': 3, 'b': 4}))
    print(res)

# 输出
[(1, 2), (3, 4)]

共享内存和进程锁

一般情况下，各个进程中的数据变量是无法发生交流的，但我们可以通过使用Value数据存储在一个共享的内存表中。

import multiprocessing as mp
import time


def job(v, num):
    for _ in range(5):
        time.sleep(0.1)  # 暂停0.1秒，让输出效果更明显
        v.value += num  # v.value获取共享变量值
        print(v.value)


def multicore():
    v = mp.Value('i', 0)  # 定义共享变量
    p1 = mp.Process(target=job, args=(v, 1))
    p2 = mp.Process(target=job, args=(v, 3))  # 设定不同的number看如何抢夺内存
    p1.start()
    p2.start()
    p1.join()
    p2.join()


if __name__ == '__main__':
    multicore()

# 输出
1
5
9
13
17
4
8
12
16
20

在上面的代码中，我们定义了一个共享变量v，两个进程都可以对它进行操作。 在job()中我们想让v每隔0.1秒输出一次累加num的结果，但是在两个进程p1和p2 中设定了不同的累加值。很明显可以看到他们发生了冲突。

为了解决上述不同进程抢共享资源的问题，我们可以用加进程锁来解决。

import multiprocessing as mp


def job(v, num, l):
    l.acquire()  # 锁住
    for _ in range(5):
        time.sleep(0.1)
        v.value += num  # 获取共享内存
        print(v.value)
    l.release()  # 释放


def multicore():
    l = mp.Lock()  # 定义一个进程锁
    v = mp.Value('i', 0)  # 定义共享内存
    p1 = mp.Process(target=job, args=(v, 1, l))  # 需要将lock传入
    p2 = mp.Process(target=job, args=(v, 3, l))
    p1.start()
    p2.start()
    p1.join()
    p2.join()


if __name__ == '__main__':
    multicore()

# 输出
3
6
9
12
15
16
17
18
19
20

需要主义的是上面可能仍然会发生冲突——p1先执行还是p2先执行的问题。为了解决这个问题我们可以在start,join中决定他们的顺序。

import multiprocessing as mp

def job(v, num, l):
    l.acquire()  # 锁住
    for _ in range(5):
        time.sleep(0.1)
        v.value += num  # 获取共享内存
        print(v.value)
    l.release()  # 释放

def multicore():
    l = mp.Lock()  # 定义一个进程锁
    v = mp.Value('i', 0)  # 定义共享内存
    p1 = mp.Process(target=job, args=(v, 1, l))  # 需要将lock传入
    p2 = mp.Process(target=job, args=(v, 3, l))
    p1.start()
    p1.join()
    p2.start()
    p2.join()

if __name__ == '__main__':
    multicore()

# 输出
1
2
3
4
5
8
11
14
17
20