线程

发表于 2019-12-07 更新于 2026-11-03 分类于 Python Python教程 Python基础本文字数： 17k 阅读时长 ≈ 31 分钟

前言

这一篇讲 threading：怎么起线程、线程间为什么天然共享数据、竞态与锁、Event / Timer / local 这些同步与工具类、子线程异常怎么处理、GIL 到底影响什么，以及现在更推荐的 ThreadPoolExecutor。
所有示例在本机 Python 3.14.10（Windows）上实跑过，输出原样贴在代码块下面。为了输出可复现，涉及并发时序的地方只打印布尔结论或排序后的结果，不打印会漂移的数字和顺序。

线程简介

一个进程里可以跑多个执行流，这些执行流就是线程，常被称作「轻量级进程」。
进程是资源分配的基本单位，线程是 CPU 调度的基本单位：同一进程内的线程共享地址空间（全局变量、打开的文件、堆上的对象），所以线程间传数据不用序列化。
一个进程默认只有一个线程，即主线程（threading.main_thread()）。进程退出时的规矩是：非守护线程（默认）会被等——解释器等它跑完才退出；守护线程（daemon=True）不等——主线程一走它就被直接带走，剩下的代码不再执行。下面「怎么起一个线程」里有两组实测对照。

与进程的对比：

维度	进程	线程
内存	独立地址空间	同进程内共享
创建开销	大（要复制 / 重新解释器）	小
通信	Pipe / Queue / Manager / 共享内存	直接读写共享对象（要加锁）
崩溃影响	一般只崩自己	一个线程踩坏内存可能带崩整个进程
吃多核	能	受 GIL 限制（CPU 密集不划算，IO 密集没问题）

两个模块：_thread 是低级接口（start_new_thread() 返回线程 id，没有对象、没有 join），threading 是对它的高级封装。日常一律用 threading；_thread 只需要知道它存在。

怎么起一个线程

import threading
import time

def worker(n):
    cur = threading.current_thread()
    time.sleep(0.05)
    print('子线程里：名字 =', cur.name, '| 是主线程吗:', cur is threading.main_thread(), '| 参数 =', n)

if __name__ == '__main__':
    print('主线程名:', threading.main_thread().name)
    print('开跑前活跃线程数:', threading.active_count())
    t = threading.Thread(target=worker, args=(42,), name='worker-1')
    print('start 前 ident:', t.ident, '| is_alive:', t.is_alive())
    t.start()
    print('start 后 is_alive:', t.is_alive())
    print('运行中的线程名:', sorted(x.name for x in threading.enumerate()))
    t.join()
    print('join 后 is_alive:', t.is_alive(), '| ident 还在:', t.ident is not None)
    print('结束时活跃线程数:', threading.active_count())

输出：

主线程名: MainThread
开跑前活跃线程数: 1
start 前 ident: None | is_alive: False
start 后 is_alive: True
运行中的线程名: ['MainThread', 'worker-1']
子线程里：名字 = worker-1 | 是主线程吗: False | 参数 = 42
join 后 is_alive: False | ident 还在: True
结束时活跃线程数: 1

start() 和 join() 各自到底做了什么：

start()：创建线程并让它从 target 那个函数开始跑。它是「开始」不是「跑完」——调用完立刻返回，主线程继续往下走，于是两个线程并发，谁先打印不固定。同一个对象只能 start() 一次，重复调用报 RuntimeError: threads can only be started once。
join()：让调用它的那个线程停在原地等，等被 join 的那个线程结束才往下走；不是「把两个线程合并」。返回值永远是 None，join(0.2) 表示最多等 0.2 秒、到点就返回（可能还没等完，要配 is_alive() 判断）；没 start() 就 join() 报 RuntimeError: cannot join thread before it is started；重复 join() 不报错；线程 join 自己会报 RuntimeError: cannot join current thread（死锁自己没意义）。
join() 等的是它那一个，不是「最先跑完的那个」：下面 B 先结束，t1.join() 仍然要等 A 跑完才返回。
等待期间当前线程是挂起的（不是空转轮询），不占 CPU，也不会因此霸着 GIL——这正是 join 能让 IO 型线程「同时等」的原因。

import threading
import time


def worker(tag, sec):
    time.sleep(sec)
    print('线程', tag, '跑完', flush=True)


def noop():
    pass


if __name__ == '__main__':
    t1 = threading.Thread(target=worker, args=('A', 0.6))
    t2 = threading.Thread(target=worker, args=('B', 0.2))
    t1.start()
    t2.start()
    print('两个 start 都调用完了，主线程立刻走到下一行', flush=True)
    t1.join()
    print('t1.join() 这时才返回（B 早就跑完了，等的是 A）', flush=True)
    print('t2.join() 的返回值:', t2.join())

    u = threading.Thread(target=noop)
    try:
        u.join()
    except RuntimeError as e:
        print('没 start 就 join ->', type(e).__name__, e)
    u.start()
    try:
        u.start()
    except RuntimeError as e:
        print('重复 start ->', type(e).__name__, e)
    u.join()
    try:
        threading.current_thread().join()
    except RuntimeError as e:
        print('主线程 join 自己 ->', type(e).__name__, e)
    print('重复 join 不报错，返回值:', u.join())

输出：

两个 start 都调用完了，主线程立刻走到下一行
线程 B 跑完
线程 A 跑完
t1.join() 这时才返回（B 早就跑完了，等的是 A）
t2.join() 的返回值: None
没 start 就 join -> RuntimeError cannot join thread before it is started
重复 start -> RuntimeError threads can only be started once
主线程 join 自己 -> RuntimeError cannot join current thread
重复 join 不报错，返回值: None

要点：

start() 之前 ident 是 None；join() 之后 is_alive() 变 False，但 ident 仍留着旧值（它只是记录，不代表线程还在）。
enumerate() / active_count() 会把主线程也算进去。

线程共享内存，所以「收集结果」最直接的方式就是往一个共享列表里写：

import threading

results = []

def worker(i):
    results.append(i * i)              # 线程共享进程内存，直接写列表就行

if __name__ == '__main__':
    ts = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
    for t in ts:
        t.start()
    for t in ts:
        t.join()
    print('共享列表拿到全部结果:', sorted(results))

输出：

1	共享列表拿到全部结果: [0, 1, 4, 9, 16]

这里不用加锁是因为 list.append 在 CPython 里是原子的（GIL 保护）；换成 x = d[k]; d[k] = x + 1 这种「读-改-写」就不安全了，见下一节。

守护线程：主线程走的时候，它要不要跟着走

先分清两件事：主线程的语句跑完 ≠ 进程退出。threading.Thread(...) 建出来的线程默认是非守护的，进程必须等它们全部结束才退出；写成 daemon=True 的就是守护线程（后台服务线程），进程一走它们就被直接带走。

非守护版（全程没有 `join`）：

import threading
import time

ready = threading.Event()          # 用来固定「子线程先打印」这件事，否则两线程同时输出会串行成一行


def job():
    print('子线程 开始', flush=True)
    ready.set()
    time.sleep(0.6)
    print('子线程 结束', flush=True)


if __name__ == '__main__':
    t = threading.Thread(target=job)                 # daemon 默认 False，即非守护
    t.start()
    ready.wait()
    print('主线程自己的代码跑完了（全程没有 join）', flush=True)
    print('主线程已经走到最后一行', flush=True)

输出：

子线程 开始
主线程自己的代码跑完了（全程没有 join）
主线程已经走到最后一行
子线程 结束

守护版（只多一个 `daemon=True`）：

import threading
import time

ready = threading.Event()          # 用来固定「子线程先打印」这件事，否则两线程同时输出会串行成一行


def job():
    print('子线程 开始', flush=True)
    ready.set()
    time.sleep(0.6)
    print('子线程 结束', flush=True)


if __name__ == '__main__':
    t = threading.Thread(target=job, daemon=True)                 # 守护线程
    t.start()
    ready.wait()
    print('主线程自己的代码跑完了（全程没有 join）', flush=True)
    print('主线程已经走到最后一行', flush=True)

输出：

1
2
3

子线程 开始
主线程自己的代码跑完了（全程没有 join）
主线程已经走到最后一行

两份输出只差最后一行「子线程结束」：非守护的被解释器等到跑完了，守护的还没睡醒进程就退了。把子线程的睡眠改成 3 秒再跑，非守护版进程要多活约 3 秒才退出，守护版 0 秒就退（本机实测）。

三个容易混的点：

「等不等」说的是进程退出这件事，跟谁先执行、跟 join 都没关系；join 对守护线程照样有效——实测 t.join() 会一直等到那个守护线程跑完，只是没人 join 时进程不会替它等。
主线程自己也是非守护的（threading.current_thread().daemon 是 False），所以交互式环境里起了非守护线程又忘了收尾，解释器会卡着不退。
守护线程随时可能被掐断，别让它做「必须写完」的事（落盘、发完最后一个请求）；真要收尾就用非守护 + join，或者用 with ThreadPoolExecutor(...)（退出 with 块时自动等）。

顺着这个名字还有三条实测出来的规则：

daemon 标志会被继承：守护线程里再 Thread(...) 且不明说时，新线程的 daemon 也是 True。
继承只是默认值，不是锁死：守护线程里显式写 daemon=False 开的线程，进程退出时照样会被等（实测那行的输出仍然打印了出来）。
被带走时是硬终止：守护线程里的 finally 不保证执行（实测主线程一结束，那段 finally 就没打印）；非守护线程的 finally 会正常执行。

别把「守护」理解反：它不是「主线程退出后保护子线程继续跑完」，规则恰好相反——守护线程跟着进程一起没。想要「脱离上级继续执行」是另一件事（Unix 里叫 detach / daemon 化，nohup、setsid），在 Python 里对应的不是线程标志，而是用 subprocess.Popen 起一个独立进程（实测父进程退出后它照样跑完），细节见《进程》的「守护进程与子进程异常」。线程本身做不到这件事：同一个进程里的线程没有「脱离进程单独活着」的可能，进程一停，所有线程一起停。

数据共享与竞态

共享是默认行为

线程之间共享进程的所有数据，全局变量不需要任何机制就能互相看到——这和《进程》里「全局变量改不动对方」正好相反。

竞态是怎么发生的

用两个事件把「读-改-写」强制拆开，就能稳定复现一次丢失更新：

import threading

balance = 0
step1 = threading.Event()
step2 = threading.Event()

def reader_first():
    global balance
    cur = balance                      # 读到 0
    step1.set()                        # 告诉对方「我已经读完了」
    step2.wait()                       # 等对方也写回
    balance = cur + 1                  # 再写回，把对方的修改覆盖掉

def reader_second():
    global balance
    step1.wait()                       # 等第一个线程先读
    cur = balance                      # 同样读到 0
    balance = cur + 1
    step2.set()
    step2.wait()                       # 等对方写回，保证输出稳定

if __name__ == '__main__':
    t1 = threading.Thread(target=reader_first)
    t2 = threading.Thread(target=reader_second)
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    print('两个线程各加 1，结果却是:', balance)
    print('原因：读-改-写 被拆开，后写的覆盖了先写的')

输出：

1 2	两个线程各加 1，结果却是: 1 原因：读-改-写被拆开，后写的覆盖了先写的

为什么 `money += 1` 不是原子的

import dis
import threading

money = 0

def add():
    global money
    money += 1

if __name__ == '__main__':
    ops = [i.opname for i in dis.get_instructions(add)
           if i.opname not in ('RESUME', 'RETURN_CONST', 'CACHE')]
    print('money += 1 的字节码:', ops)
    print('读和写是两条不同指令:', 'LOAD_GLOBAL' in ops and 'STORE_GLOBAL' in ops)
    print('中间被抢走就会丢更新，所以 GIL 不等于线程安全')

输出：

1
2
3

money += 1 的字节码: ['LOAD_GLOBAL', 'LOAD_SMALL_INT', 'BINARY_OP', 'STORE_GLOBAL', 'LOAD_CONST', 'RETURN_VALUE']
读和写是两条不同指令: True
中间被抢走就会丢更新，所以 GIL 不等于线程安全

字节码里 LOAD_GLOBAL 和 STORE_GLOBAL 是两条指令，GIL 只保证「一条字节码执行期间不被抢走」，不保证「这两条之间没人插队」。所以 GIL 不等于线程安全。

加锁修正

import threading

money = 0
lock = threading.Lock()

def add(n, times):
    global money
    for _ in range(times):
        with lock:                     # 自动 acquire / release
            money += 1

if __name__ == '__main__':
    ts = [threading.Thread(target=add, args=(1, 20000)) for _ in range(4)]
    for t in ts:
        t.start()
    for t in ts:
        t.join()
    print('4 个线程各加 20000 次，结果 =', money, '| 期望 =', 4 * 20000)
    print('一致吗:', money == 80000)

输出：

1 2	4 个线程各加 20000 次，结果 = 80000 \| 期望 = 80000 一致吗: True

with lock: 会自动 acquire / release；手写 lock.acquire() 时务必 try/finally，否则中途抛异常就永久死锁。

同步原语

`Lock` 与 `RLock`

Lock 不可重入：同一线程二次 acquire() 会自锁死。RLock（可重入锁）允许同一线程嵌套获取，常用于「公开方法加锁、内部方法也加锁」的类。先把方法逐个说清（Lock / RLock 是同一套接口）：

acquire()：抢锁。空闲就抢到并返回 True；锁被别人占着就就地等。两个参数可以改这个行为：acquire(blocking=False) 不等待，抢不到立刻返回 False；acquire(timeout=0.1) 最多等 0.1 秒，等不到返回 False（实测确实只等了 0.11 秒）。
release()：归还锁。锁本来没人拿着还去 release，会报 RuntimeError: release unlocked lock，所以别自己手写配平，用 with lock: 让它在退出时自动释放（中途抛异常也能释放）。
locked()：问一句「现在被占着吗」，返回布尔值。只能拿来调试——「先判断没被占、再去抢」这两步之间随时可能被人抢走，不能当互斥用。
RLock（可重入锁）：同一线程可以嵌套 acquire，套几层就要 release 几次，计数回到 0 才真正放开；对不同线程之间它和 Lock 一样互斥。适合「公开方法已经加锁、它调用的内部方法也加同一把锁」的类。
为什么要有锁：now += 1 其实是「读、算、写」三步，两个线程交错执行就会丢更新（上一节有实测）。锁的作用是把那段代码变成「同一时刻只有一个线程在里面」。

import threading

rlock = threading.RLock()

def outer():
    with rlock:
        inner()
        return 'ok'

def inner():
    with rlock:                        # 同一把锁再进一次，普通 Lock 会自锁死
        return 'inner'

if __name__ == '__main__':
    print('可重入锁嵌套获取:', outer())
    print('当前持有者:', threading.get_ident() == rlock._owner if hasattr(rlock, '_owner') else '释放后无持有者')

输出：

1 2	可重入锁嵌套获取: ok 当前持有者: 释放后无持有者

`Semaphore`：限制并发数

Semaphore(n) 不是锁，是一个许可计数器（签名 Semaphore(value=1)）：

acquire()：拿一个许可，计数减 1；已经是 0 就阻塞等别人归还。同样支持 acquire(blocking=False)（拿不到返回 False）和 acquire(timeout=...)。
release()：归还一个许可，计数加 1。它不检查你有没有拿过——凭空 release() 会把计数加到超过初始值（实测 Semaphore(1) 先 release 一次，后面能连拿两个）。
要严格不许超发就用 BoundedSemaphore(n)，多 release 会报 ValueError: Semaphore released too many times。
和 Lock 的区别：锁讲「谁持有」，信号量讲「还剩几个名额」；Semaphore(1) 行为上像一把锁，但用错语义（多 release）不会被拦住。
下面这段里的 lock 只是为了安全地改 now/peak 两个计数器，限流是 sem 干的，两个对象各司其职。

import threading
import time

sem = threading.Semaphore(2)           # 最多 2 个线程同时进入
lock = threading.Lock()
now = 0
peak = 0

def job(i):
    global now, peak
    with sem:
        with lock:
            now += 1
            peak = max(peak, now)
        time.sleep(0.05)
        with lock:
            now -= 1

if __name__ == '__main__':
    ts = [threading.Thread(target=job, args=(i,)) for i in range(6)]
    for t in ts:
        t.start()
    for t in ts:
        t.join()
    print('6 个任务，实际最大并发 =', peak, '| 上限 =', 2)

输出：

1	6 个任务，实际最大并发 = 2 \| 上限 = 2

典型用途：同时最多 N 个线程访问数据库 / 调外部接口。

`Event`：跨线程发信号

Event 是一个开关，让线程「等一个条件成立」，它不保护数据（签名 Event()，没有参数）：

新建时默认是关的：is_set() 返回 False。
set()：打开开关，所有正在 wait() 的线程立刻被放行；已经是开的再 set 没有副作用。
clear()：关上开关，下一批 wait() 又要重新等。
wait(timeout=None)：开关是开的就立刻返回 True；关着就阻塞。超时也会返回，返回值就是「有没有等到」——实测关着时 wait(0.1) 等 0.11 秒返回 False，所以别把 wait() 当无返回值的 sleep 用。
和 Condition 的分工：Event 是「广播一个状态」，谁都能看、谁都能等；Condition 要配锁用（with cond: 里 wait()/notify()），适合「等到某个条件为真再继续」。没持锁就 notify() 会报 RuntimeError: cannot notify on un-acquired lock（实测）。

import threading
import time

e = threading.Event()

def worker():
    for i in range(3):
        print('worker 等待指令...')
        ok = e.wait(timeout=2)         # 超时也会返回，返回值就是是否等到
        print('等到信号:', ok, '-> 执行第', i + 1, '步')
        e.clear()                      # 复位，下一轮还要等

if __name__ == '__main__':
    t = threading.Thread(target=worker)
    t.start()
    for i in range(3):
        time.sleep(0.05)
        print('主线程发出第', i + 1, '次信号')
        e.set()
    t.join()
    print('结束')

输出：

worker 等待指令...
主线程发出第 1 次信号
等到信号: True -> 执行第 1 步
worker 等待指令...
主线程发出第 2 次信号
等到信号: True -> 执行第 2 步
worker 等待指令...
主线程发出第 3 次信号
等到信号: True -> 执行第 3 步
结束

set() 唤醒所有 wait() 的线程，clear() 之后下一轮要重新等；wait(timeout=) 会返回布尔值，别写无限等待。还有两个同族：Condition（配合 wait / notify 做「等待某个条件成立」，生产者-消费者经典写法）、Barrier（N 个线程互相等齐再一起走）。

定时线程与线程局部数据

`Timer`：延迟执行

Timer(interval, function, args=None, kwargs=None) 是延时执行的线程：它是 Thread 的子类，所以 start()/join()/is_alive() 那套全都在（实测 isinstance(t, threading.Thread) 为 True）。

建完不会自己跑，必须 start()；没 start 时 is_alive() 是 False。
cancel()：在时间到之前调用可以取消，函数就不会执行（实测取消后那次回调没发生）；已经跑起来了就取消不了。
默认不是守护线程（t.daemon 是 False，3.14 实测），所以主线程退出前会等它——这点和很多人印象里的「Timer 是 daemon」不同，别靠记忆下结论。
它只保证「不早于 interval 执行」，具体晚多少看调度；要周期执行得在函数里再建一个 Timer，或者用 schedule 这类库。

import threading
import time

def greet(name):
    print('定时器到点，你好', name)

if __name__ == '__main__':
    t = threading.Timer(0.2, greet, args=('小明',))
    start = time.perf_counter()
    t.start()
    print('start 之后主线程不用等:', time.perf_counter() - start < 0.1)
    t.join()
    print('任务只执行一次:', not t.is_alive())
    cancel_timer = threading.Timer(10, greet, args=('不会被打印',))
    cancel_timer.start()
    cancel_timer.cancel()              # 到点前取消
    print('cancel 之后 is_alive:', cancel_timer.is_alive())

输出：

start 之后主线程不用等: True
定时器到点，你好 小明
任务只执行一次: True
cancel 之后 is_alive: True

Timer 是 Thread 的子类，start() 后在后台等 N 秒再执行；cancel() 只能在到点之前取消——所以取消之后 is_alive() 仍然是 True（线程还在那儿等超时），只是任务不会执行了。

`threading.local`：每个线程一份

threading.local() 造出来的是一个按线程分格子的对象：同一个属性名，每个线程各存各的，互不干扰。

每个线程第一次访问时都是空的：读没写过的属性报 AttributeError: '_thread._local' object has no attribute 'v'（实测）。
子线程写的值，主线程读不到，另一个子线程也读不到——这正是它的全部用途：给每个线程一份私有上下文。
常见场景：每线程一个数据库连接、请求上下文、线程内计数器。
它解决的是「隔离」，不是「并发安全」：同一线程内不需要锁，跨线程共享的那份还是要 Lock。

import threading
import time

local = threading.local()
results = {}
lock = threading.Lock()


def show(tag):
    local.value = tag                    # 每个线程各有一份，互不覆盖
    time.sleep(0.02)                     # 让三个线程真正同时在跑
    with lock:
        results[tag] = local.value


if __name__ == '__main__':
    ts = [threading.Thread(target=show, args=('线程%d' % i,)) for i in range(3)]
    for t in ts:
        t.start()
    for t in ts:
        t.join()
    for tag in sorted(results):          # 排序后再打印，输出才稳定
        print(tag, '->', results[tag])
    print('主线程里读 local:', getattr(local, 'value', '没设置过'))

输出：

线程0 -> 线程0
线程1 -> 线程1
线程2 -> 线程2
主线程里读 local: 没设置过

同一个对象，不同线程读写互不干扰，常用来放「当前请求上下文」「数据库连接」。注意它不是「自动按线程清理」：线程结束后其数据随线程对象回收，线程池里复用线程时要自己重置。

自定义线程类与子线程异常

继承 Thread 重写 run()，start() 会自动调用它：

import threading
import time

class MyThread(threading.Thread):

    def __init__(self, delay, times):
        super().__init__()
        self.delay = delay
        self.times = times

    def run(self):                     # start() -> 自动调用 run()
        for _ in range(self.times):
            print('子线程中...')
            time.sleep(self.delay)

if __name__ == '__main__':
    t = MyThread(0.02, 3)
    t.start()
    t.join()
    print('自定义线程类跑了', t.times, '轮 | 线程名:', t.name)

输出：

子线程中...
子线程中...
子线程中...
自定义线程类跑了 3 轮 | 线程名: Thread-1

子线程里抛异常不会让主线程崩，默认只在 stderr（专门输出错误的那条通道，和 print 走的 stdout 分开，终端里看起来是同一个地方）打一段 traceback（就是报错时那一大段「从哪层调用进来」的记录）。想统一处理，换 threading.excepthook：

import threading

def boom():
    raise ValueError('子线程炸了')

if __name__ == '__main__':
    caught = []
    old = threading.excepthook

    def hook(args):
        caught.append(args.exc_value)

    threading.excepthook = hook
    t = threading.Thread(target=boom)
    t.start()
    t.join()
    print('excepthook 抓到异常:', type(caught[0]).__name__, caught[0])
    threading.excepthook = old
    print('主进程还活着')

输出：

1 2	excepthook 抓到异常: ValueError 子线程炸了主进程还活着

GIL 到底影响什么

CPython 的全局解释器锁（GIL）保证同一时刻只有一个线程在执行 Python 字节码。结论分两类：

CPU 密集：多线程不会变快（甚至更慢，因为切换开销），要吃多核请用 multiprocessing 或 ProcessPoolExecutor。
IO 密集：线程在等待（网络、磁盘、sleep）时会释放 GIL，所以多线程照样有效。

import threading
import time

def cpu_job(n):
    x = 0
    for i in range(n):
        x += i
    return x

def io_job(sec):
    time.sleep(sec)

if __name__ == '__main__':
    N = 1_500_000
    t0 = time.perf_counter()
    cpu_job(N)
    cpu_job(N)
    seq_cpu = time.perf_counter() - t0

    t0 = time.perf_counter()
    ts = [threading.Thread(target=cpu_job, args=(N,)) for _ in range(2)]
    for t in ts:
        t.start()
    for t in ts:
        t.join()
    par_cpu = time.perf_counter() - t0

    t0 = time.perf_counter()
    ts = [threading.Thread(target=io_job, args=(0.2,)) for _ in range(4)]
    for t in ts:
        t.start()
    for t in ts:
        t.join()
    par_io = time.perf_counter() - t0

    print('CPU 密集：多线程并没有更快 ->', par_cpu >= seq_cpu * 0.8)
    print('IO 密集：4 个 sleep(0.2) 并发完成，总耗时 <', 0.5, '秒 ->', par_io < 0.5)

输出：

1 2	CPU 密集：多线程并没有更快 -> True IO 密集：4 个 sleep(0.2) 并发完成，总耗时 < 0.5 秒 -> True

补充两点：调用 C 扩展（用 C 语言写的模块，比如 numpy 的底层，执行时不跑 Python 字节码）时若显式释放 GIL 也能并行；3.13 起有实验性的 free-threaded 构建（彻底去掉 GIL），但生态还在跟进，生产慎用。

现代写法：ThreadPoolExecutor

手搓 Thread + Lock + 结果列表的活，交给线程池更省事：

import concurrent.futures as cf
import threading
import time

names = set()
lock = threading.Lock()

def io_task(n):
    time.sleep(0.05)                   # 模拟 IO：等待时会释放 GIL
    with lock:
        names.add(threading.current_thread().name)
    return n * 2

if __name__ == '__main__':
    with cf.ThreadPoolExecutor(max_workers=4) as ex:
        print('map 保序:', list(ex.map(io_task, range(6))))
        futures = [ex.submit(io_task, i) for i in range(3)]
        print('submit + result:', [f.result() for f in futures])
    print('用到的线程数不超过 max_workers:', len(names) <= 4)

输出：

1
2
3

map 保序: [0, 2, 4, 6, 8, 10]
submit + result: [0, 2, 4]
用到的线程数不超过 max_workers: True

map 保序返回；submit 给 Future，.result(timeout=...) 取值，子线程里的异常会在这里原样抛出——这是它比裸 Thread 好用的最大一点。
with 退出时自动 shutdown(wait=True)。
想「谁先完成先处理」用 concurrent.futures.as_completed(futures)。
同一个 Executor 换 ProcessPoolExecutor 就能从线程切到进程，接口一致，这也是判断「该用线程还是进程」最省事的实验方式。

整体示例：并发抓取 + 计数

把线程、锁、共享字典串起来：6 个线程各自「抓取」，成功失败分开计数，共享结构全部在锁里改。

import threading
import time

urls = ['a', 'b', 'c', 'd', 'e', 'f']
lock = threading.Lock()
done = []
counter = {'ok': 0, 'fail': 0}

def fetch(url):
    time.sleep(0.03)                   # 模拟网络请求
    ok = url in ('a', 'c', 'e')        # 一半成功一半失败
    with lock:                         # 共享结构必须加锁
        if ok:
            counter['ok'] += 1
            done.append(url)
        else:
            counter['fail'] += 1

if __name__ == '__main__':
    ts = [threading.Thread(target=fetch, args=(u,)) for u in urls]
    for t in ts:
        t.start()
    for t in ts:
        t.join()
    print('抓取成功（排序后）:', sorted(done))
    print('统计:', counter)
    print('总数对得上:', counter['ok'] + counter['fail'] == len(urls))

输出：

1
2
3

抓取成功（排序后）: ['a', 'c', 'e']
统计: {'ok': 3, 'fail': 3}
总数对得上: True

done 的追加顺序取决于调度，所以打印前先 sorted()——并发程序里凡是顺序相关的输出，要么排序、要么带序号，否则日志没法比对。

前言

线程简介

怎么起一个线程

守护线程：主线程走的时候，它要不要跟着走

非守护版（全程没有 join）：

守护版（只多一个 daemon=True）：