Python 中如何实现多线程？
python约 5 分钟接口没挂，CPU 也不高，导出任务就是慢。
日志里每个用户的订单明细都要调一次外部接口，单次 300ms 左右，串起来跑 200 个用户，时间直接奔着一分钟去了。这种场景我一般不会先去抠算法，先看它到底是在算，还是在等。
像这种日志，基本就很明显：
2026-06-04 10:18:21 fetch user=10021 cost=318ms
2026-06-04 10:18:22 fetch user=10022 cost=291ms
2026-06-04 10:18:22 fetch user=10023 cost=337ms
...
每一行都在等网络返回。Python 多线程在这种地方就有用了。
别一听 Python 多线程就先扯 GIL。GIL 确实存在，CPU 密集型任务，比如压缩大文件、跑复杂计算、批量图片处理，多线程不一定快，甚至可能更慢。但如果你的程序大部分时间都在等接口、等数据库、等磁盘，那线程切出去干别的活，是能省时间的。
最直接的写法是 threading.Thread。
import time
import random
import threadingdefpull_user_bill(user_id: int):
start = time.time()
# 模拟一次远程接口调用
time.sleep(random.uniform(0.2, 0.5))
cost = int((time.time() - start) * 1000)
print(f"[bill-sync] user={user_id} cost={cost}ms")
users = [10021, 10022, 10023, 10024, 10025]
threads = []
for uid in users:
t = threading.Thread(
target=pull_user_bill,
args=(uid,),
name=f"bill-worker-{uid}"
)
t.start()
threads.append(t)
for t in threads:
t.join()
print("all user bill pulled")
这里有两个点别省。
一个是 start()，线程真正开始跑。
一个是 join()，主线程等子线程全部跑完。你要是忘了 join()，脚本类任务里很容易出现一种现象：主流程提前结束，后面的结果还没来得及处理。这个坑不大，但挺烦。
不过上面这种写法，我只会在临时脚本里用。线上代码如果这么写，一批 5000 个用户，你就敢创建 5000 个线程，机器不骂人才怪。
线程不是免费的。每个线程都有栈空间，线程太多之后，CPU 光在线程切换上就够喝一壶。这个地方要加控制。
更像生产代码的写法，是用队列。
import time
import queue
import random
import threadingtask_box = queue.Queue(maxsize=200)
bad_users = []
deffetch_score(user_id: int) -> int:
time.sleep(random.uniform(0.1, 0.4))
if user_id % 17 == 0:
raise TimeoutError("score service timeout")
return random.randint(1, 100)
defworker():
whileTrue:
user_id = task_box.get()
if user_id isNone:
task_box.task_done()
break
try:
score = fetch_score(user_id)
print(f"[score-ok] user={user_id}, score={score}")
except Exception as e:
bad_users.append(user_id)
print(f"[score-fail] user={user_id}, err={e}")
finally:
task_box.task_done()
workers = []
for i in range(8):
t = threading.Thread(target=worker, name=f"score-worker-{i}")
t.start()
workers.append(t)
for uid in range(10001, 10101):
task_box.put(uid)
for _ in workers:
task_box.put(None)
task_box.join()
for t in workers:
t.join()
print("failed users:", bad_users)
这个写法丑一点，但我更信。
Queue 负责塞任务，8 个 worker 负责消费。线程数固定，不会因为任务多就把机器打爆。None 是停止信号，告诉线程可以收工了。
这里我故意把失败用户放到了 bad_users 里。实际业务里别只打印日志，最好把失败数据落下来，后面补偿。多线程最怕那种“看起来跑完了，其实中间丢了几条”的代码。
还有一个问题，多个线程同时改同一个变量，会出事。
比如你想统计成功数量，很多人会这么写：
success_count += 1
这行看着像一步，其实不是。读出来、加一、写回去，中间都可能被别的线程插进来。数据量小的时候看不出来，一上量就开始少几个，查起来很恶心。
要么别共享变量，要么加锁。
import threadingstat_lock = threading.Lock()
stat = {
"ok": 0,
"fail": 0
}
defmark_ok():
with stat_lock:
stat["ok"] += 1
defmark_fail():
with stat_lock:
stat["fail"] += 1
锁别乱加。能只锁两行，就别把整个接口调用包进去。外部接口本来就慢，你把锁套在外面，8 个线程又被你写回了单线程。
平时我更常用的是 ThreadPoolExecutor，代码短，也不容易写散。
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import randomdefcheck_sku_stock(sku_id: str):
time.sleep(random.uniform(0.1, 0.35))
if sku_id.endswith("9"):
raise RuntimeError("stock api rejected")
return sku_id, random.randint(0, 50)
sku_list = [f"sku-{i}"for i in range(30)]
with ThreadPoolExecutor(max_workers=6, thread_name_prefix="stock") as pool:
futures = {
pool.submit(check_sku_stock, sku): sku
for sku in sku_list
}
for future in as_completed(futures):
sku = futures[future]
try:
sku_id, stock = future.result()
print(f"[stock] {sku_id}={stock}")
except Exception as e:
print(f"[stock-error] sku={sku}, err={e}")
这个版本适合大多数接口并发、文件批处理、批量校验任务。
max_workers 不要拍脑袋写 100、200。IO 慢一点可以开大一些，但也要看下游扛不扛得住。你本地跑得飞快，下游库存服务被你打挂了，那不叫优化，那叫制造事故。
我一般会先从 4、8、16 这种小数字试起，看日志里的平均耗时和失败率。如果失败率开始上升，或者下游开始报限流，就别再加线程了。
多线程还有个细节：异常不会自动抛到主线程。
直接用 threading.Thread 时，子线程里炸了，主线程可能还在那儿继续跑。日志没打好，你甚至不知道哪条数据失败。ThreadPoolExecutor 的好处是 future.result() 会把异常重新抛出来，至少你能集中处理。
最后说下 Python 多线程适合放在哪些地方。
批量调接口，适合。
批量读小文件，适合。
批量查数据库，要谨慎，连接池别被打满。
CPU 计算，不太适合，优先看多进程或者换实现方式。
多线程不是让代码“高级”的东西，它就是个工具。看到程序慢，先别急着上线程，先把日志打出来，看时间到底耗在哪里。要是时间都花在等待上，多线程能救；要是时间都花在计算上，它大概率帮不上忙。
这一步判断错了，后面代码写得再漂亮也没用。