企业官网建设流程全解析

前言

标题里的「最新推荐」其实是个模糊说法——Python 并发不存在一个放之四海皆准的最新推荐。选哪种模型，取决于任务是 CPU 密集还是 I/O 密集，取决于要不要跨平台，取决于能不能接受进程间的序列化成本。本文不讲机制原理（那是另一篇文章的事），而是给出三套可以直接复制运行的典型骨架，让你拿去改成自己的任务。

每套骨架都遵循同样的工程约定：用concurrent.futures或asyncio的高层接口而不是裸 API、有明确的错误处理、有资源清理。文中提到的ProcessPoolExecutor在 Windows 下的约束会单独点明，因为这是跨平台最容易翻车的地方。

所有示例以 Python 3 为基准，代码块首行标注了适用版本。

一、多进程典型骨架：CPU 密集

CPU 密集任务（大量计算、图像处理、压缩）要绕过 GIL 用多核，就得开进程。推荐用ProcessPoolExecutor，它比裸multiprocessing.Pool更现代，也更好和with语句配合。

# 适用于 Python 3.8+；Windows 下必须放在 if __name__ 保护内调用
from concurrent.futures import ProcessPoolExecutor, as_completed
import math

def heavy(n):
# 纯计算，放进子进程，不受 GIL 限制
return sum(math.sqrt(i) for i in range(n))

def main():
jobs = [1_000_000, 1_200_000, 800_000, 900_000]
results = []
with ProcessPoolExecutor(max_workers=4) as pool:
futures = {pool.submit(heavy, n): n for n in jobs}
for fut in as_completed(futures):
n = futures[fut]
try:
results.append((n, fut.result()))
except Exception as exc: # 子进程里的异常会在这里重新抛出
print(f"任务 {n} 失败: {type(exc).__name__}")
return results

if __name__ == "__main__": # Windows 用 spawn，必须有这个保护
for n, total in main():
print(n, round(total, 2))

要点：

• 
  
• pool.submit(fn, arg)返回Future，as_completed让你按完成先后处理，不必等最慢的。
• 
  
• fut.result()会把子进程里发生的异常重新抛出到主进程，所以要包try/except。
• 
  
• 必须有if __name__ == "__main__":：Windows 用 spawn 启动子进程，子进程会重新导入模块，没有保护会无限递归创建进程。
• 
  
• 参数与返回值必须可 pickle。上面传的是普通函数和整数，没问题；如果传 lambda、数据库连接就会报错。
• 
  

二、多线程典型骨架：I/O 密集

网络请求、文件读写这类等待多的任务，用线程池就够。GIL 会在线程等待 I/O 时释放，所以多个线程能真正重叠起来。

# 适用于 Python 3.8+
from concurrent.futures import ThreadPoolExecutor, as_completed
import urllib.request

def download(url, timeout=10):
req = urllib.request.Request(url, headers={"User-Agent": "demo-bot/1.0"})
with urllib.request.urlopen(req, timeout=timeout) as resp:
return url, len(resp.read())

def main(urls, workers=8):
out = {}
with ThreadPoolExecutor(max_workers=workers) as pool:
futures = {pool.submit(download, u): u for u in urls}
for fut in as_completed(futures):
url = futures[fut]
try:
_, size = fut.result()
out[url] = size
except Exception as exc:
out[url] = f"ERROR: {type(exc).__name__}"
return out

if __name__ == "__main__":
result = main(["https://example.com/"] * 6, workers=4)
print(len(result))

要点：

• 
  
• max_workers别设太大。线程数超过目标站能接受的频率，只会更快被限流。
• 
  
• 每个请求都设timeout，否则一个卡住的连接会一直占着一个线程。
• 
  
• 线程池的任务函数必须是线程安全的：如果多个任务写同一个字典，要么改成一任务一结果、最后合并，要么加锁。
• 
  

三、协程典型骨架：高并发 I/O

当连接数很多（几百上千），线程的栈开销和切换成本开始变得可观，这时候协程合适。协程在单线程里靠await让位，能把「等待」大量重叠。

# 适用于 Python 3.8+
import asyncio

async def worker(name, delay, sem):
async with sem: # 限制同时进行的任务数
await asyncio.sleep(delay) # 模拟 I/O 等待
return f"{name} 完成"

async def main(names):
sem = asyncio.Semaphore(3)
tasks = [asyncio.create_task(worker(n, 0.1, sem)) for n in names]
results = await asyncio.gather(*tasks, return_exceptions=True)
for r in results:
print(r)

if __name__ == "__main__":
asyncio.run(main([f"job{i}" for i in range(10)]))

要点：

• 
  
• 先create_task把任务全部排队，再gather统一等待——写成一串await就退化成串行了。
• 
  
• gather(..., return_exceptions=True)让单个任务失败不会中断整组。
• 
  
• Semaphore用来封顶并发数，避免一次性打开过多连接。
• 
  
• 协程里绝对不能出现time.sleep或同步 HTTP 客户端，那会把事件循环堵死。
• 
  

四、同一件事的三种写法

假设要把一批数据分别处理，下面这张表帮你快速选型：

混合场景要拆开：用协程并发下载，把下载结果交给ProcessPoolExecutor做计算。硬塞进同一个模型里，要么被 GIL 卡住，要么被进程间序列化拖慢。

五、骨架的公共部分：资源清理与异常处理

三套骨架看起来品种不同，但工程上要处理的问题几乎一样，把这几点做成习惯，代码的健壮性会明显不同。

第一，用上下文管理器管池子。with ProcessPoolExecutor(...) as pool:和with ThreadPoolExecutor(...) as pool:在退出时会调用shutdown(wait=True)，等所有已提交任务跑完并回收工作进程/线程。如果手动创建却忘了关，进程可能残留、线程可能泄漏。协程这边对应的是async with aiohttp.ClientSession()这类写法。

第二，异常不要吞掉。线程池和进程池都会把任务里的异常保存到Future里，只有调用fut.result()时才重新抛出。如果你只用as_completed拿结果却不调用result()，异常就被静默丢弃了——任务失败了你却什么都不知道。所以每个fut.result()都该包在try/except里，并记录是哪个任务失败的。

第三，失败要能定位。上面的骨架都用{fut: 任务标识}的字典把Future映射回原始参数，这样报错时能说清「哪个 URL / 哪个 n 出错了」，而不是甩一个孤零零的异常。这一点在批量任务里尤其重要。

第四，超时是必需的，不是可选的。ThreadPoolExecutor没有全局超时开关，超时必须落在每个任务函数内部（urlopen(..., timeout=...)）；asyncio这边可以用asyncio.wait_for给单个任务加时间上限。没有超时，一个卡死的任务就能把整个批次拖成「永远跑不完」。

第五，注意max_workers的默认值。ThreadPoolExecutor若不指定，默认值随版本变过好几次：3.5 起是 CPU 数乘 5，3.8 起改成min(32, os.cpu_count() + 4)，3.13 又调整过。与其记住这些规则，不如显式写出你要的并发数，让行为可预期。进程池同理，显式写max_workers更方便调优和排查。

# 适用于 Python 3.8+；带统一异常收集的线程池骨架
from concurrent.futures import ThreadPoolExecutor, as_completed

def run_batch(items, fn, workers=4):
ok, failed = [], []
with ThreadPoolExecutor(max_workers=workers) as pool:
futures = {pool.submit(fn, it): it for it in items}
for fut in as_completed(futures):
item = futures[fut]
try:
ok.append((item, fut.result()))
except Exception as exc:
failed.append((item, type(exc).__name__))
return ok, failed

这个run_batch就是骨架的「可复用内核」：把任务函数换掉，三种模型都能套。区别只在于把ThreadPoolExecutor换成ProcessPoolExecutor，或者换成asyncio.gather的协程版本。

常见坑点

1. 
   
2. Windows 上多进程忘了if __name__ == "__main__":
3. 
   

❌ 顶层直接pool.submit(...)，子进程导入时反复创建

✅ 把所有进程池启动代码放进if __name__ == "__main__":块

1. 
   
2. 给进程池传不可 pickle 的对象
3. 
   

❌ 传 lambda、文件对象、数据库连接

✅ 只传可序列化的数据，连接在子进程里现建

1. 
   
2. 把 CPU 密集任务塞进线程池
3. 
   

❌ 期待ThreadPoolExecutor跑满多核

✅ CPU 密集换ProcessPoolExecutor

1. 
   
2. 协程里混用阻塞调用
3. 
   

❌async def f(): time.sleep(1)

✅await asyncio.sleep(1)

1. 
   
2. 线程池任务写同一份共享状态不加锁
3. 
   

❌ 多线程同时shared[k] = v又同时读

✅ 一任务一结果再合并，或用threading.Lock

1. 
   
2. 不给任务设超时
3. 
   

❌urlopen(url)无 timeout，一个卡死连接占住线程

✅urlopen(url, timeout=10)

1. 
   
2. max_workers/ 并发数设成几百上千
3. 
   

❌ 以为越多越快

✅ 从个位数起步，按目标站与服务端反馈调整

总结

三套骨架的结构其实高度相似：建池 / 建任务、提交、按完成顺序收集、逐任务兜异常。真正的差别只在「任务卡在 CPU 还是卡在等待」。拿到新需求时先问自己这一句，选型基本就不会错。