多进程编程

multiprocessing

由于 GIL(全局解释锁) 的原因, 多线程并不能充分利用多核处理器, 如果是一个 CPU 计算型的任务, 应该使用多进程模块 multiprocessing, 它的工作方式与线程库不同, 但是两种库的接口相似。multiprocessing 给每个进程赋予了单独的 Python 解释器, 这样就规避了 GIL 所带来的问题。

import multiprocessing


def worker():
    print('Worker')


if __name__ == '__main__':
    jobs = []
    for i in range(5):
        p = multiprocessing.Process(target=worker)
        jobs.append(p)
        p.start()

# 输出:
Worker
Worker
Worker
Worker
Worker

目标函数也支持传入参数:

import multiprocessing


def worker(num):
    print(f'Worker: {num}')


if __name__ == '__main__':
    jobs = []
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(i, ))
        jobs.append(p)
        p.start()


# 输出:
Worker: 0
Worker: 1
Worker: 3
Worker: 2
Worker: 4

守护进程

和多线程模块一样, 多进程也是可以设置守护进程的, 守护进程可以一直运行而不影响主程序的结束。

import multiprocessing
import time


def daemon():
    p = multiprocessing.current_process() # 获取当前的进程
    print(f'Starting: {p.name} {p.pid}')
    time.sleep(2)
    print(f'Exiting: {p.name} {p.pid}')



def non_daemon():
    p = multiprocessing.current_process()
    print(f'Starting: {p.name} {p.pid}')
    print(f'Exiting: {p.name} {p.pid}')


if __name__ == '__main__':
    d = multiprocessing.Process(
        name = 'daemon',
        target = daemon,
        daemon=True
    )

    n = multiprocessing.Process(
        name = 'non-daemon',
        target = non_daemon,
    )

    d.start()
    time.sleep(1)
    n.start()

# 输出:
Starting: daemon 7068
Starting: non-daemon 10116
Exiting: non-daemon 10116

通过上面代码的输出可以看到, 守护进程没有完成程序就结束了。如果需要等待守护进程完成工作后再结束, 可以使用 join() 方法:

import multiprocessing
import time


def daemon():
    p = multiprocessing.current_process() # 获取当前的进程
    print(f'Starting: {p.name} {p.pid}')
    time.sleep(2)
    print(f'Exiting: {p.name} {p.pid}')



def non_daemon():
    p = multiprocessing.current_process()
    print(f'Starting: {p.name} {p.pid}')
    print(f'Exiting: {p.name} {p.pid}')


if __name__ == '__main__':
    d = multiprocessing.Process(
        name = 'daemon',
        target = daemon,
        daemon=True
    )

    n = multiprocessing.Process(
        name = 'non-daemon',
        target = non_daemon,
    )

    d.start()
    time.sleep(1)
    n.start()

    d.join()
    n.join()

# 输出:
Starting: daemon 2928
Starting: non-daemon 13376
Exiting: non-daemon 13376
Exiting: daemon 2928

join() 方法接收一个超时的参数, 默认是 none, 表示会一直阻塞, 可以设置一个超时时间, 在这个时间内没有完成, 就会阻塞:

import multiprocessing
import time


def daemon():
    p = multiprocessing.current_process() # 获取当前的进程
    print(f'Starting: {p.name} {p.pid}')
    time.sleep(2)
    print(f'Exiting: {p.name} {p.pid}')



def non_daemon():
    p = multiprocessing.current_process()
    print(f'Starting: {p.name} {p.pid}')
    print(f'Exiting: {p.name} {p.pid}')


if __name__ == '__main__':
    d = multiprocessing.Process(
        name = 'daemon',
        target = daemon,
        daemon=True
    )

    n = multiprocessing.Process(
        name = 'non-daemon',
        target = non_daemon,
    )

    d.start()
    time.sleep(1)
    n.start()

    d.join(1) # 传入的是 1 秒, 小于守护进程里面设置的两秒
    print('d.is_alive()', d.is_alive()) # 得到进程当前的状态
    n.join()

# 输出:
Starting: daemon 7272
Starting: non-daemon 5872
Exiting: non-daemon 5872
d.is_alive() True

进程池

任务的执行周期决定了 CPU 核数和任务的分配算法, 使用多进程编程 Pool 是一个很灵活的保证效率的方法:

from functools import lru_cache
from multiprocessing import Pool


# lur_cache 装饰器使用最近最少使用算法
# 会把最近使用的对象存储起来, 并把最近最少使用的对象在缓存值达到预设值之前从内存中移除
# 这里 maxsize 设置为 None, 表示永远都不会达到这个预设值
# lru_cache 这个装饰器适合把耗时的函数的执行结果保存起来, 避免传入相同的参数时重复计算
@lru_cache(maxsize=None)
def fib(n):
    if n < 2:
        return n
    return fib(n-1) + fib(n-2)

if __name__ == '__main__':
    # 在 windows 下运行会造成 RuntimeError, 应该使用 __name__ == '__main__' 来保护程序的入口点
    pool = Pool(2)
    pool.map(fib, [35] * 2)

dummy

multiprocessing.dummy 这个子模块虽然在多进程模块的代码中, 但是接口和多线程的接口基本是一样的, 如果分不清一个任务是 CPU 密集型还是 I/O 密集型, 可以使用如下方法去试:

from multiprocessing import Pool
from multiprocessing.dummy import Pool

这种兼容的方式, 便于在多线程/多进程之间切换。

Queue(队列)

多线程里面有 Queue 模块实现队列, 多进程的 multiprocessing 里面包含了 Queue 这个类, 它是线程和进程安全的。下面是一个生产者/消费者的例子, 用到了两个队列, 一个队列用于存储完成的任务, 另外一个用于存储任务完成后的结果:

import time
from multiprocessing import Process, JoinableQueue, Queue
from random import random


def double(n):
    return n * 2


def producer(in_queue):
    while 1:
        wait_time = random()
        time.sleep(wait_time)
        in_queue.put((double, wait_time))
        if wait_time > 0.9:
            in_queue.put(None)
            print('停止生产')
            break


def consumer(in_queue, out_queue):
    while 1:
        task = in_queue.get()
        if task is None:
            break
        func, arg = task
        result = func(arg)
        in_queue.task_done()
        out_queue.put(result)


if __name__ == '__main__':
    tasks_queue = JoinableQueue() # JoinableQueue 里面有 join() 方法和 task_done() 方法, 而 Queue 内没有, tasks_queue 可以用这两个方法来标识任务是否完成以决定要不要阻塞下去
    results_queue = Queue()
    processes = []

    p = Process(target=producer, args=(tasks_queue, ))
    p.start()
    processes.append(p)

    p = Process(target=consumer, args=(tasks_queue, results_queue))
    p.start()
    processes.append(p)

    tasks_queue.join()

    for p in processes:
        p.join()

    while True:
        if results_queue.empty():
            break
        result = results_queue.get()
        print(f'Result: {result}')

# 输出:
停止生产
Result: 1.3500119015795484
Result: 1.7651301976930043
Result: 1.6336519677702004
Result: 0.06429843269363
Result: 0.29352347406759494
Result: 1.0097954936153397
Result: 0.19863644698178606
Result: 0.9589181928209678
Result: 1.4618869426710388
Result: 1.6837862156424794
Result: 0.8653351112396082
Result: 1.5958573192798793
Result: 0.15849993035736087
Result: 1.3471427672620973
Result: 1.7492282062851205
Result: 0.27695109993667644
Result: 0.7201581558818728
Result: 1.9614106580291402

进程间共享状态

multiprocessing 提供了在进程之间共享状态的方案, 主要有两种: 共享内存 和 服务器进程。

共享内存

共享内存主要通过 Value 和 Array 来实现, 在多个进程之间共享一份数据, 常见的共享类型有下面这些:

>>> from multiprocessing.sharedctypes import typecode_to_type
>>> typecode_to_type
{'c': <class 'ctypes.c_char'>, # 左边是缩写, 右边是全称
'u': <class 'ctypes.c_wchar'>,
'b': <class 'ctypes.c_byte'>,
'B': <class 'ctypes.c_ubyte'>,
'h': <class 'ctypes.c_short'>,
'H': <class 'ctypes.c_ushort'>,
'i': <class 'ctypes.c_long'>,
'I': <class 'ctypes.c_ulong'>,
'l': <class 'ctypes.c_long'>,
'L': <class 'ctypes.c_ulong'>,
'f': <class 'ctypes.c_float'>,
'd': <class 'ctypes.c_double'>}

下面是一个共享内存的实现示例:

from multiprocessing import Process, Lock
from multiprocessing.sharedctypes import Value, Array
from ctypes import Structure, c_bool, c_double


class Point(Structure):
    _fields_ = [('x', c_double), ('y', c_double)]


def modify(n, b, s, arr, A):
    n.value **= 2
    b.value = True
    s.value = s.value.upper()
    arr[0] = 10
    for a in A:
        a.x **= 2
        a.y **= 2


if __name__ == '__main__':
    lock = Lock()

    n = Value('i', 7) # 这里使用的共享类型的缩写
    b = Value(c_bool, False, lock=False)
    s = Array('c', b'hello world', lock=lock)
    arr = Array('i', range(5), lock=True)
    A = Array(Point, [(1.525, -6.25), (-5.75, 2.5)], lock=lock)

    p = Process(target=modify, args=(n, b, s, arr, A)) # 通过 modify() 把这些内容的值更改
    p.start()
    p.join()

    print(n.value)
    print(b.value)
    print(s.value)
    print(arr[:])
    print([(a.x, a.y) for a in A])

# 输出
49
True
b'HELLO WORLD'
[10, 1, 2, 3, 4]
[(2.3256249999999996, 39.0625), (33.0625, 6.25)]

服务器进程

一个 multiprocessing 的 Manager 对象会控制一个服务器的进程, 其他进程可以通过代理的方式来访问这个服务器进程, 常见的共享方式有以下几种:

Namespace: 创建一个可分享的命名空间。
Value/Array: 和共享内存中共享 ctypes 对象的方式一样。
dict/list: 创建一个可分享的 dict/list, 支持对应数据结构的方法。
Condition/Event/Lock/Queue/Semaphore: 创建一个可分享的对应同步原语的对象。

from multiprocessing import Manager, Process


def modify(ns, lproxy, dproxy):
    ns.a **= 2
    lproxy.extend(['b', 'c'])
    dproxy['b'] = 1


if __name__ == '__main__':
    manager = Manager()
    ns = manager.Namespace() # 创建一个命名空间
    ns.a = 1
    lproxy = manager.list()
    lproxy.append('a')
    dproxy = manager.dict()
    dproxy['b'] = 0

    p = Process(target=modify, args=(ns, lproxy, dproxy))
    p.start()
    print(f'PID: {p.pid}')
    p.join()

    print(ns.a)
    print(lproxy)
    print(dproxy)

# 输出:
PID: 6556
1
['a', 'b', 'c']
{'b': 1}

分布式的进程间通信

使用 manager 可以实现一个简单的分布式的不同服务器之间不同进程的通信, 也就是一个简单的 client/server 模型:

# remote_server.py

from multiprocessing.managers import BaseManager


class RemoteManager(BaseManager):
    pass


if __name__ == '__main__':
    host = '127.0.0.1'  # 主机名
    port = 5000 # 端口
    authkey = b'secret' # 验证 key

    shared_list = [] # 分享列表

    RemoteManager.register('get_list', callable=lambda: shared_list) # 远程服务器支持 get_list 方法, 可以获得 shared_list 的值
    mgr = RemoteManager(address=(host, port), authkey=authkey)
    server = mgr.get_server()
    server.serve_forever()

# client.py

from multiprocessing.managers import BaseManager


class RemoteManager(BaseManager):
    pass


if __name__ == '__main__':
    host = '127.0.0.1'
    port = 5000
    authkey = b'secret'

    RemoteManager.register('get_list')
    mgr = RemoteManager(address=(host, port), authkey=authkey)
    mgr.connect()

    l = mgr.get_list() # 拿到分享的 list
    print(l) # 一开始 l 为 空
    l.append(1) # 添加一个值 1
    print(mgr.get_list())

> python remote_server.py

> python client.py
[]
[1]

多进程编程