一行 Python 代码实现程序并行

那天晚上快十一点，我在公司楼下便利店门口蹲着吃饭，手机连着公司 Wi-Fi 在远程跑脚本，心想：要是这玩意儿能快点就好了。那是个很简单的 Python 脚本，几十个接口要挨个调一遍，结果主线程老老实实一个一个来，CPU 都没怎么动，时间倒是咔咔往上加。

后来我就顺手改了一行代码，脚本立马像打了鸡血一样，十几秒搞定。那一瞬间就一个感觉：并行这东西，别被“高大上”这三个字给吓住了。

我今天就用这个小故事，把“一行 Python 代码让程序并行起来”这事儿，从头捋一遍。

先说清楚：那“一行”到底是哪一行？

别被标题骗了啊，啥叫“一行实现并行”？不是说整个程序只有一行，而是说：真正让任务“并行跑起来”的，就那一行关键代码。

比如我那个调接口的脚本，最原始的长这样：

import requests
deffetch(url):
    resp = requests.get(url, timeout=5)
return url, resp.status_code
urls = [
"https://example.com/api/1",
"https://example.com/api/2",
# ... 一大堆
]
results = []
for u in urls:
    results.append(fetch(u))

这个版本，最慢的地方其实就是那几行 for 循环，所有请求乖乖排队，一个没回来，下一个不许上车。

后来我就把最后那几行，换成了这一行：

from concurrent.futures import ThreadPoolExecutor
results = list(ThreadPoolExecutor().map(fetch, urls))

逻辑没变，参数也没变，但从“一个一个”变成了“一群人一起干活”。真正起决定作用的，就是这一句 ThreadPoolExecutor().map(...)，就是我们说的“一行并行”。

为什么用线程池这一招？

你想想，大部分业务脚本里面，那种跑得特别慢的地方，通常有两类：

• 要么是一直在算：比如压缩视频、图片处理、加密解密；
• 要么是一直在等：比如等网络、等磁盘、等数据库。

等的这类，就很适合用线程池来一块儿干。

上面那行代码拆开说，其实就三件事：

from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=8) as pool:
    results = list(pool.map(fetch, urls))

只是我偷懒写成了一行：

results = list(ThreadPoolExecutor(max_workers=8).map(fetch, urls))

• ThreadPoolExecutor(max_workers=8)：开了一个最多 8 条线程的线程池；
• .map(fetch, urls)：把 urls 这个列表丢进去，让每个线程都去调 fetch；
• list(...)：把结果收回来，变成一个列表。

对我当时那个场景来说，就是一行代码，把单线程串行 I/O，变成了 8 个线程并发 I/O。

给你一个能跑的完整小例子

免得说得太虚，我写个能直接复制跑的，还是用 requests 模拟一下接口调用：

import time
from concurrent.futures import ThreadPoolExecutor
import requests
deffetch(url):
    resp = requests.get(url, timeout=5)
return url, resp.status_code
defserial(urls):
    start = time.time()
    results = [fetch(u) for u in urls]
    print("串行耗时：", time.time() - start)
return results
defparallel(urls):
    start = time.time()
# 就这一行，把串行改成并行
    results = list(ThreadPoolExecutor(max_workers=8).map(fetch, urls))
    print("并行耗时：", time.time() - start)
return results
if __name__ == "__main__":
    urls = ["https://httpbin.org/delay/1"] * 16# 每个请求都延迟 1 秒
    serial(urls)
    parallel(urls)

这个脚本跑起来，大概是：

• 串行版本差不多 16 秒；
• 并行版本 8 个线程，大概 2~3 秒搞定。

你细品一下，就改了一行循环的写法，体感完全不一样。

顺便把 GIL 那点事说清楚

很多同学一听“Python 多线程”，下意识就会冒出一句：Python 有 GIL 啊，线程没用。

这个说法不完全对。

简单粗暴的区分法：

• IO 密集型（主要在等）：网络请求、读写磁盘、数据库操作 → 用 ThreadPoolExecutor 这种多线程，效果很好；
• CPU 密集型（主要在算）：大循环、加密、压缩、图像运算 → 线程会被 GIL 卡脖子，得上多进程 ProcessPoolExecutor。

比如下面这个“一行并行求平方”的例子，用多进程比较合适：

from concurrent.futures import ProcessPoolExecutor
nums = list(range(1_000_000))
defheavy(x):
return x * x
results = list(ProcessPoolExecutor().map(heavy, nums))

同样也是一行 .map(...)，只是把线程池换成了进程池，背后多开了几个独立的 Python 进程，每个都有自己的 GIL，CPU 能吃满。

如果你要写完整一点，大概是这样：

from concurrent.futures import ProcessPoolExecutor
import time
defheavy(x: int) -> int:
    s = 0
for _ in range(100_000):
        s += x * x
return s
defrun_parallel():
    nums = list(range(10))
    start = time.time()
with ProcessPoolExecutor() as pool:
        results = list(pool.map(heavy, nums))
    print("结果：", results)
    print("并行耗时：", time.time() - start)
if __name__ == "__main__":
    run_parallel()

里面真正“放飞”的，还是这一句：

results = list(ProcessPoolExecutor().map(heavy, nums))

一行并行背后的几个小坑，得提前说

你看代码写得少了，坑一般都在“看不见”的地方。

我踩过几个，给你先打个预防针。

1）Windows 要记得那句保护

多进程的时候，在 Windows 上一定要把并行的那一行，放到：

if __name__ == "__main__":
    ...

里面，不然就会出现进程无限自我复制那种诡异问题。

错的写法（会炸）：

from concurrent.futures import ProcessPoolExecutor
defheavy(x): ...
nums = list(range(10))
# 可能在 import 时就把子进程再 fork 一遍，死循环那种
results = list(ProcessPoolExecutor().map(heavy, nums))

对的写法：

from concurrent.futures import ProcessPoolExecutor
defheavy(x): ...
nums = list(range(10))
if __name__ == "__main__":
    results = list(ProcessPoolExecutor().map(heavy, nums))

2）别偷懒用大对象做参数

那天我们组的小李，把一个几百 MB 的大对象当参数丢进 ProcessPoolExecutor，然后跟我吐槽“并行怎么还更慢了”。

进程之间传参数，要序列化、拷贝的，你参数越大，花在“搬东西”上的时间越多，CPU 真正算的时间就相对变少了。

比较靠谱的写法是：

• 参数上尽量传“小东西”：数字、短字符串、ID；
• 大对象放到进程里自己读，比如从文件/数据库里按 ID 查。

3）线程池大小别乱写

max_workers 那个参数，很多人随手写个 100，看着挺爽，其实经常是浪费。

粗暴一点的经验值：

• IO 密集型：max_workers 可以写成 CPU 核数的几倍，比如 4 核机子，线程开到 20、30 也还好；
• CPU 密集型：一般就写成 os.cpu_count() 或者 os.cpu_count() + 1。

比如这样：

import os
from concurrent.futures import ThreadPoolExecutor
pool = ThreadPoolExecutor(max_workers=os.cpu_count() * 5)

再来一个“极致懒”的一行用法

有时候你甚至连函数都懒得单独定义，就写个 lambda 也行，虽然不太推荐，但偶尔写脚本图快，可以这么干：

from concurrent.futures import ThreadPoolExecutor
urls = ["https://httpbin.org/delay/1"] * 8
results = list(
    ThreadPoolExecutor().map(
lambda u: (u, __import__("requests").get(u).status_code),
        urls
    )
)

这一坨压起来其实还是那句 .map(...) 在发力，只不过函数是临时拼出来的。线上代码别这么写，自己看着都累。

还有几个一行并行的兄弟招式

除了 concurrent.futures，Python 里还有一些“看起来就很一行”的并行库，顺口说两个，你以后遇到不至于陌生。

1）multiprocessing.Pool

这个算是老牌选手了：

from multiprocessing import Pool
with Pool(4) as p:
    results = p.map(heavy, nums)

一行并行依旧是 p.map(heavy, nums) 这一句，思路跟 ProcessPoolExecutor 很像，只是接口风格不太一样。

2）joblib.Parallel（数据科学圈常客）

很多做机器学习的同学会见到这种：

from joblib import Parallel, delayed
results = Parallel(n_jobs=4)(delayed(heavy)(x) for x in nums)

这里真正“啪的一下开并行”的，也是 Parallel(...)(...) 那一行。你仔细比对，会发现大家核心思想都是一样的：给你一个函数、一堆参数，帮你并行 map 一下。

最后随手给你一个通用小模板

你以后如果写脚本，遇到“有一堆任务长得一模一样，只是参数不一样”的场景，脑子里就可以直接套这个模板：

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
defwork(x):
# 这里写你的业务逻辑
return x * x
defrun_parallel(data, use_process=False, workers=None):
    Executor = ProcessPoolExecutor if use_process else ThreadPoolExecutor
with Executor(max_workers=workers) as pool:
# 一行搞定并行
return list(pool.map(work, data))
if __name__ == "__main__":
    nums = list(range(20))
# IO 密集：用线程
    results1 = run_parallel(nums, use_process=False, workers=8)
    print("线程并行：", results1)
# CPU 密集：用进程
    results2 = run_parallel(nums, use_process=True)
    print("进程并行：", results2)

里面真正值得你记住的，其实就一句：

list(Executor(max_workers=workers).map(work, data))

剩下那些 import、函数封装、参数传递，都是在给这一行铺垫舞台。

行，差不多就聊到这儿，我去给自己冲杯咖啡，你要是正好有个脚本跑得特别慢，不妨回去就试试把那个 for 循环，悄悄换成一行 .map()，感受一下“并行”这俩字的分量。