Python多进程开发之如何轻松突破GIL瓶颈_Python

1. 为啥要用多进程？先认清gil

python多线程为何没法加速cpu密集型任务？因为gil（全局解释器锁）：在主流的cpython解释器中，无论多少线程归根结底同时只有一个线程在执行python字节码，严重影响cpu计算性能的并行。

多进程机制完美绕开了gil：每个进程拥有自己的python解释器和内存空间，真正多核、多cpu并行执行，cpu密集型任务大大提速！

2. python多进程基础用法：multiprocessing库

python 内置标准库 multiprocessing，api与threading非常类似，上手容易。

2.1 proces介绍

process([group [, target [, name [, args [, kwargs]]]]])

target 表示调用对象
args 表示调用对象的位置参数元组
kwargs 表示调用对象的字典
name 为别名
group 实质上不使用

2.2 hello，多进程

from multiprocessing import process
import os

def worker():
    print('子进程pid:', os.getpid())

if __name__ == '__main__':
    p = process(target=worker)
    p.start()
    p.join()
    print('主进程结束')

注意！ windows下多进程要写在 if __name__ == '__main__': 保护块内，否则会无限递归启动进程。

2.3 多进程并行计算

from multiprocessing import process
import time

def task(name):
    print(f'{name} 开始')
    time.sleep(2)
    print(f'{name} 结束')

if __name__ == '__main__':
    proc_list = []
    for i in range(3):
        p = process(target=task, args=(f'进程{i+1}',))
        p.start()
        proc_list.append(p)
    for p in proc_list:
        p.join()
    print('所有子进程结束')

多个进程可并发分担任务，真正多核利用！

2.4 子进程类实现（面向对象）

import multiprocessing
import time

class clockprocess(multiprocessing.process):
    def __init__(self, interval):
        multiprocessing.process.__init__(self)
        self.interval = interval

    def run(self):
        n = 5
        while n > 0:
            print("当前时间: {0}".format(time.ctime()))
            time.sleep(self.interval)
            n -= 1


if __name__ == '__main__':
    p = clockprocess(3)
    p.start()

3. 进阶：进程池批量并行(process pool)

大量任务并发时，手动管理进程很麻烦，用pool池自动分配和收集结果，极大提升效率！

from multiprocessing import pool
import os, time

def square(n):
    print('计算', n, '在进程', os.getpid())
    time.sleep(1)
    return n * n

if __name__ == '__main__':
    with pool(processes=3) as pool:
        results = pool.map(square, range(5))
    print('返回结果:', results)

比循环创建 process 简洁得多，还能自动分配cpu，每个子进程并发执行给定任务。

4. 多进程间数据：通信与共享

每个进程独立内存，不能像多线程那样直接共享变量
多进程用队列（queue）、管道（pipe）或特殊共享对象实现

4.1 用队列 queue 通信（推荐）

from multiprocessing import process, queue

def producer(q):
    for i in range(5):
        q.put(i)
    q.put(none)  # 发送结束标记

def consumer(q):
    while true:
        item = q.get()
        if item is none:
            break
        print('消费:', item)

if __name__ == '__main__':
    q = queue()
    p1 = process(target=producer, args=(q,))
    p2 = process(target=consumer, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

4.2 共享内存变量：value 与 array

from multiprocessing import process, value, array

def worker(val, arr):
    val.value += 10
    for i in range(len(arr)):
        arr[i] *= 2

if __name__ == '__main__':
    num = value('i', 5)                # int类型共享变量
    arr = array('i', [1, 2, 3, 4])     # 数组
    p = process(target=worker, args=(num, arr))
    p.start()
    p.join()
    print('num =', num.value)
    print('arr =', list(arr))

5. 多进程安全：锁机制

多进程也需防止竞争条件，通过 lock/manager 加锁同步。

from multiprocessing import process, lock, value

def add(lock, counter):
    for _ in range(100000):
        with lock:
            counter.value += 1

if __name__ == '__main__':
    lock = lock()
    counter = value('i', 0)
    procs = [process(target=add, args=(lock, counter)) for _ in range(4)]
    for p in procs:
        p.start()
    for p in procs:
        p.join()
    print('结果:', counter.value)