Python 并发编程

发表于 分类于
从线程创建与同步讲到GIL计时实验与多进程，梳理Python并发编程的选型思路

本文把线程、锁、队列、GIL 和多进程串成一条线：先用 threading 跑起最小的多线程程序，再用锁和队列解决共享数据与返回值的问题，接着用计时实验看清 GIL 对计算密集型任务的限制，最后换成 multiprocessing 绕开它，并给出选型建议。

1. 线程基础

1.1 创建并启动线程

threading 是标准库提供的线程模块。创建线程只要把要执行的函数交给 Thread(target=...)，再调用 start() 让它开始工作。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import threading


def thread_job():
    print('This is a thread of %s' % threading.current_thread())


def thread_info():
    print(threading.active_count())    # 已激活的线程数
    print(threading.enumerate())       # 所有线程的信息
    print(threading.current_thread())  # 当前正在运行的线程


if __name__ == '__main__':
    thread_info()
    thread = threading.Thread(target=thread_job)  # 定义线程
    thread.start()                                # 让线程开始工作

注意入口写法是 if __name__ == '__main__':，前后各两个下划线。左边变量名写错（比如多打一个下划线的 __name___）反倒好查：那是个没定义过的全局名，求值时当场 NameError: name '__name___' is not defined，脚本直接崩。真正难缠的是右边字符串写错，if __name__ == '__main___': 语法没问题、运行也不报错，只是条件永远为假，整段代码被静默跳过——这类笔误在并发脚本里特别难察觉。

1.2 用 join 等待线程结束

start() 之后主线程会继续往下跑，不会等子线程。join() 的作用是让主线程阻塞在这里，直到对应的子线程结束。下面 T1 要睡 1 秒，T2 立刻结束，加上 join() 之后 all done 一定是最后打印的。

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
import threading
import time


def t1_job():
    print('T1 start')
    for _ in range(10):
        time.sleep(0.1)
    print('T1 finish')


def t2_job():
    print('T2 start')
    print('T2 finish')


def main():
    t1 = threading.Thread(target=t1_job, name='T1')
    t2 = threading.Thread(target=t2_job, name='T2')
    t1.start()
    t2.start()
    t1.join()   # 主线程在这里等 T1 结束
    t2.join()   # 再等 T2 结束
    print('all done')


if __name__ == '__main__':
    main()

2. 线程同步

2.1 不加锁：输出会互相插队

两个线程同时修改同一个全局变量 A，谁先执行、执行到哪一步都不确定，print 的输出会互相插队。

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
import threading

A = 0


def job1():
    global A
    for _ in range(10):
        A += 1
        print('job1', A)


def job2():
    global A
    for _ in range(10):
        A += 10
        print('job2', A)


if __name__ == '__main__':
    t1 = threading.Thread(target=job1)
    t2 = threading.Thread(target=job2)
    t1.start()
    t2.start()
    t1.join()
    t2.join()

打印结果很杂乱，两个线程的输出被切碎混在了一起：

1
2
3
4
5
6
7
8
9
job1 job21
11job1
 job212
22job1
job2  3323

job2job1  4344

job2job1  5554

这里要说清一件事：这段示例演示的是 print 输出交错，并不是数据竞争。print 一次调用会分两次写流（先写内容再写换行），中间可能被切走，所以字符会碎；但 A 的值本身并没有丢。

想验证这一点，把 print 去掉、把循环放大，看最终结果对不对：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import threading, sys

A = 0

def worker():
    global A
    for _ in range(1_000_000):
        A += 1

sys.setswitchinterval(1e-9)          # 把切换间隔压到极限，逼它出错
threads = [threading.Thread(target=worker) for _ in range(4)]
[t.start() for t in threads]
[t.join() for t in threads]
print(A)      # 4000000，一次都没丢

我在 CPython 3.14 上跑过：4 个线程各累加 100 万次，sys.setswitchinterval 压到 1e-9，结果精确等于 4000000，丢更新一次都复现不出来。原因见下面 GIL 那一节——A += 1 编译出的几条字节码之间没有切换检查点。

要真造出丢更新，得在”读”和”写”之间插进一次函数调用，制造一个检查点：

1
2
3
4
5
6
7
8
def ident(x):
    return x

def worker():
    global A
    for _ in range(200_000):
        t = ident(A)      # 读完就有机会被切走
        A = t + 1         # 回来时 A 可能已经被别人改过

同样 4 个线程，期望 800000，实测只有 383817——丢了 41 万多次。这才是真正的数据竞争，也才是下一节 Lock 要解决的问题。

2.2 用 Lock 保护共享数据

Lock 保证同一时刻只有一个线程能进入被保护的代码段：lock.acquire() 上锁，lock.release() 释放。推荐直接用 with lock:，异常时也能自动释放，不会把锁漏在半路上。

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
import threading

A = 0
lock = threading.Lock()


def job1():
    global A
    with lock:            # 等价于 lock.acquire() ... lock.release()
        for _ in range(10):
            A += 1
            print('job1', A)


def job2():
    global A
    with lock:
        for _ in range(10):
            A += 10
            print('job2', A)


if __name__ == '__main__':
    t1 = threading.Thread(target=job1)
    t2 = threading.Thread(target=job2)
    t1.start()
    t2.start()
    t1.join()
    t2.join()

这次两段循环各自完整执行，结果干净且可预期：

1
2
3
4
5
6
7
8
job1 1
job1 2
...
job1 10
job2 20
job2 30
...
job2 110

2.3 用 Queue 收集线程的运算结果

被 Thread 调用的函数没法用 return 把值交回主线程，常见做法是把结果放进队列。queue.Queue 本身是线程安全的，比自己加锁维护一个共享列表省心。

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
import threading
from queue import Queue


def job(data, q):
    """对列表的每个元素求平方，把结果放进队列"""
    q.put([x ** 2 for x in data])


def multithreading():
    q = Queue()          # q 中存放返回值，代替 return
    data = [[1, 2, 3], [3, 4, 5], [4, 4, 4], [5, 5, 5]]
    threads = []

    for item in data:
        # target 只写函数名（不加括号），参数放在 args 里
        t = threading.Thread(target=job, args=(item, q))
        t.start()
        threads.append(t)

    for t in threads:
        t.join()         # 逐个 join 回主线程

    results = [q.get() for _ in data]
    print(results)


if __name__ == '__main__':
    multithreading()

需要留意的是，q.get() 取出的顺序取决于各线程完成的先后，和 data 的顺序不一定一致。如果结果必须对应上原始输入，就把索引一起 put 进队列。

3. GIL 与计算密集型任务

前面的例子里线程都在 sleep 或者做很轻的活儿，看不出问题。一旦换成纯计算，多线程就露馅了：CPython 有一把 GIL（Global Interpreter Lock，全局解释器锁），同一时刻只允许一个线程执行 Python 字节码。要分清它限制的是”并行”而不是”并发”——即使全是纯计算、一次 I/O 都没有，解释器也会按 sys.getswitchinterval()（默认 5 ms）周期性地把 GIL 交出去，几个计算线程照样交替往前推进，看上去是”同时在跑”的。真正的代价在于它们永远凑不到同一时刻一起执行，总吞吐不会因为线程变多而增加，多个 CPU 核心也就跑不满。

这里值得往下挖一层：GIL 的交还并不是发生在每两条字节码之间。解释器维护一个 “eval breaker” 标志，只在特定指令上才去检查它——主要是 JUMP_BACKWARD（循环回边）、CALL、RESUME 这类。也就是说，一段不含调用、不含回边的直线字节码序列是不会被打断的。

这一条能一次解释两件事。第一，为什么纯计算的线程也会交替推进：for 循环每转一圈都要经过 JUMP_BACKWARD，那里就是检查点。第二，为什么上面 2.1 里 A += 1 的丢更新复现不出来——它编译成：

1
2
3
4
5
6
7
8
9
10
11
import dis

def f():
    global A
    A += 1

dis.dis(f)
#   LOAD_GLOBAL       0 (A)
#   LOAD_SMALL_INT    1
#   BINARY_OP        13 (+=)
#   STORE_GLOBAL      0 (A)

这四条之间一个检查点都没有，整个”读—改—写”是原子完成的；检查点落在它们之后的 JUMP_BACKWARD 上，那时值已经写回去了。一旦像 2.1 末尾那样插进一次 ident(A) 调用，CALL 就把检查点塞进了读和写中间，竞态立刻出现。

所以”+= 不是原子操作，所以要加锁”这个说法，结论对但推理不完整：真正决定会不会出问题的是读写之间有没有落进检查点，而这依赖于具体编译出的字节码。靠”看起来是一行代码”来判断安全性是不可靠的——该加锁还是要加锁，只是别指望用 A += 1 把 bug 演示出来。

3.1 用上下文管理器给整段实验计时

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
import threading
import time
from contextlib import contextmanager


@contextmanager
def cost_time(label):
    start = time.time()
    print('%s start : %s' % (label, time.strftime('%Y-%m-%d %H:%M:%S')))
    yield
    print('%s cost  : %.2f s' % (label, time.time() - start))


def decrement(n):
    while n > 0:
        n -= 1


TOTAL = 100_000_000

if __name__ == '__main__':
    # 单线程：一个线程做完 TOTAL 次递减
    with cost_time('single thread'):
        decrement(TOTAL)

    # 双线程：两个线程各做一半，总工作量与上面完全相同
    with cost_time('two threads'):
        t1 = threading.Thread(target=decrement, args=(TOTAL // 2,))
        t2 = threading.Thread(target=decrement, args=(TOTAL // 2,))
        t1.start()
        t2.start()
        t1.join()
        t2.join()

两次的总工作量都是 1 亿次递减，唯一的区别是由一个线程做还是两个线程分着做——只有保证工作量相同，比较才有意义。

3.2 换成装饰器写法：小心把计时器装错位置

同样的实验也可以用装饰器来计时，但要装饰整个实验，而不是装饰 decrement 本身。如果把 @cost_time 加在 decrement 上，两个子线程会各自打印自己那一半的耗时，既拿不到总耗时，也就没有了和单线程对照的意义。

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
import functools
import threading
import time


def cost_time(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        print('%s cost : %.2f s' % (func.__name__, time.time() - start))
        return result
    return wrapper


def decrement(n):
    while n > 0:
        n -= 1


@cost_time
def single_thread(total):
    decrement(total)


@cost_time
def two_threads(total):
    t1 = threading.Thread(target=decrement, args=(total // 2,))
    t2 = threading.Thread(target=decrement, args=(total // 2,))
    t1.start()
    t2.start()
    t1.join()
    t2.join()


if __name__ == '__main__':
    single_thread(100_000_000)
    two_threads(100_000_000)

3.3 实验结论

跑下来会发现：双线程版本完全没有变快，耗时和单线程基本持平，有时还略慢一些。原因就是 GIL 把两个线程的计算强行串行化了，核心数再多也用不上，多出来的那点时间花在线程切换和锁的争夺上。具体差多少和解释器版本有关——较新的 CPython 切换代价低，往往表现为持平；越老的版本越容易看到明显变慢。

所以对计算密集型任务来说，Python 的多线程确实是鸡肋；它真正的用武之地是 I/O 密集型场景——线程在等网络、等磁盘时会释放 GIL，其他线程才有机会推进。

4. 多进程 multiprocessing

multiprocessing 用来弥补 threading 的这个短板。每个进程有独立的解释器和独立的 GIL，因此可以真正并行地吃满多核。

4.1 和 threading 几乎一致的用法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import multiprocessing
import threading


def job_t(a, b):
    print('thread', a, b)


def job_p(a, b):
    print('process', a, b)


if __name__ == '__main__':
    t1 = threading.Thread(target=job_t, args=(1, 2))
    p1 = multiprocessing.Process(target=job_p, args=(1, 2))
    t1.start()
    p1.start()
    t1.join()
    p1.join()

Process 的接口和 Thread 几乎一模一样，但必须写 if __name__ == '__main__': 守卫：在 Windows 和 macOS 上子进程以 spawn 方式启动，会重新导入主模块，缺了守卫，子进程一边导入一边又执行到 p.start()。好在现在的 multiprocessing 自带防护，不会真的失控刷出一堆进程，而是在子进程里直接抛 RuntimeError: An attempt has been made to start a new process before the current process has finished its bootstrapping phase，把脚本拦下来。

4.2 进程间通信用 multiprocessing.Queue

进程之间不共享内存，queue.Queue 只在同一进程的线程之间有效，跨进程要用 multiprocessing.Queue。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import multiprocessing as mp


def job(q):
    res = 0
    for i in range(1000):
        res += i + i ** 2 + i ** 3
    q.put(res)


if __name__ == '__main__':
    q = mp.Queue()
    p1 = mp.Process(target=job, args=(q,))
    p2 = mp.Process(target=job, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    res1 = q.get()
    res2 = q.get()
    print(res1, res2, res1 + res2)

4.3 普通、多线程、多进程的效率对比

把同一份计算任务分别用三种方式跑一遍，工作量完全相同（都是两轮一百万次的累加）。

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
47
48
49
50
51
52
53
54
55
56
57
58
59
60
import functools
import multiprocessing
import threading
import time


def cost_time(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        print('%s cost time : %.4f s' % (func.__name__, time.time() - start))
        return result
    return wrapper


def job(q):
    res = 0
    for i in range(1000000):
        res += i + i ** 2 + i ** 3
    q.put(res)


@cost_time
def normal():
    res = 0
    for _ in range(2):
        for i in range(1000000):
            res += i + i ** 2 + i ** 3
    print('normal:', res)


@cost_time
def multithread():
    q = multiprocessing.Queue()
    t1 = threading.Thread(target=job, args=(q,))
    t2 = threading.Thread(target=job, args=(q,))
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    print('multithread:', q.get() + q.get())


@cost_time
def multicore():
    q = multiprocessing.Queue()
    p1 = multiprocessing.Process(target=job, args=(q,))
    p2 = multiprocessing.Process(target=job, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    print('multicore:', q.get() + q.get())


if __name__ == '__main__':
    normal()
    multithread()
    multicore()

打印结果：

1
2
3
4
5
6
normal: 499999666667166666000000
normal cost time : 0.8630 s
multithread: 499999666667166666000000
multithread cost time : 1.8855 s
multicore: 499999666667166666000000
multicore cost time : 0.4704 s

这份数据是早年在老解释器上留下的，多线程被拖慢了一倍多，线程切换的开销在当时相当可观。同一段代码在 Python 3.14 上重跑：

1
2
3
normal      cost time : 0.2138 s
multithread cost time : 0.2274 s
multicore   cost time : 0.1385 s

耗时排序还是 多进程 < 普通 < 多线程，但多线程只比普通版慢 6% 左右，和上一节递减实验”基本持平、有时略慢”的结论对得上——新版解释器的切换代价低了很多，可吞吐一点没涨，GIL 还在那儿摆着。多进程则把两份计算真正摊到了两个核上，耗时接近对折，这一点新老版本都一样。

5. 选型建议

  • I/O 密集型（网络请求、文件读写、数据库查询）：用多线程。线程等待 I/O 时会释放 GIL，并发效果明显，而且线程比进程轻得多。
  • 计算密集型（数值计算、图像处理、加解密）：用多进程。每个进程独立持有 GIL，能吃满多核；代价是启动慢、内存开销大，数据要经过序列化才能在进程间传递。
  • 共享数据：只要有多个线程写同一份数据就必须上锁，能用队列传结果时优先用队列——它本身线程安全，比自己维护锁更不容易出错。
  • 两种方案的骨架几乎一样（Thread / Process + start + join），所以可以先写多线程版本，发现瓶颈在 CPU 上再换成多进程，改动量很小。

5.1 “计算密集型别用多线程”这条结论正在失效

上面第二条是在有 GIL 的前提下才成立的。PEP 703 之后 CPython 提供了自由线程（free-threading）构建：3.13 以 python3.13t 的形式作为实验特性给出，3.14 起转为正式支持的构建配置。在这种解释器上 GIL 可以被真正关掉，多线程做纯计算能吃满多核。

1
2
import sys
sys._is_gil_enabled()     # free-threading 构建下关掉 GIL 时返回 False

需要注意的是它不是免费的：单线程性能有一定损失，而且 C 扩展必须显式声明支持（不支持的扩展一被导入就会把 GIL 重新打开）。所以现阶段的判断依据变成了”你的依赖链有没有跟上”，而不再是”Python 多线程不能算数”。

5.2 收返回值：ThreadPoolExecutor 比手写 Queue 省事

前面 2.3 用 Queue 收结果，遇到了”取出顺序和输入顺序不一致”的问题。concurrent.futures 直接把这件事解决掉了：

1
2
3
4
5
6
7
from concurrent.futures import ThreadPoolExecutor

def job(x):
    return x * x

with ThreadPoolExecutor(max_workers=4) as pool:
    print(list(pool.map(job, range(1, 5))))     # [1, 4, 9, 16]，保序

executor.map 的结果按输入顺序返回，不用自己排；想要”谁先完成先处理”则用 submit 拿 Future，配 as_completed 遍历。异常也不会像裸线程那样默默消失在子线程里，而是在 Future.result() 时重新抛给你。