異步編程進階：asyncio、threading 和多進程在實戰中的選擇
作者：用戶007
在Python并發編程中，沒有絕對的最優方案，只有最適合當前場景的方案。理解GIL的影響、掌握三種方案的特點和適用場景，選擇合適的方案進行組合使用，這才是實戰中的最佳實踐。
Python中的并發編程一直是開發者的難題。asyncio提供了異步編程的原生支持，threading提供了多線程能力，multiprocessing提供了多進程支持。這三種方案各有所長，在不同的場景中發揮不同的作用。
本文將基于真實的應用場景，深入分析這三種并發方案的原理、性能和最佳實踐，幫助你在實戰中做出正確的選擇。
三種并發模型的本質區別
Python的全局解釋器鎖（GIL）是理解三種并發模型的關鍵：
┌─────────────────────────────────────┐
│ Python解釋器 │
│ ┌──────────────────────────────┐ │
│ │ GIL(全局鎖) │ │
│ │ 一次只允許一個線程執行 │ │
│ └──────────────────────────────┘ │
└─────────────────────────────────────┘
asyncio：單線程，在I/O等待時切換任務 ? 不受GIL影響
threading：多線程，但受GIL限制 ? CPU密集型無法并行
multiprocessing：多進程，每個進程獨立GIL ? CPU密集型可并行方案一：asyncio - I/O密集型的最優選擇
asyncio的工作原理：
import asyncio
asyncdeffetch_data(url):
"""模擬獲取數據"""
print(f"開始獲取 {url}")
await asyncio.sleep(2) # 模擬I/O操作
print(f"完成獲取 {url}")
returnf"Data from {url}"
asyncdefmain():
# 并發執行多個異步任務
tasks = [
fetch_data("http://example.com/1"),
fetch_data("http://example.com/2"),
fetch_data("http://example.com/3"),
]
results = await asyncio.gather(*tasks)
return results
# 運行
results = asyncio.run(main())
# 耗時：約2秒（并發），而不是6秒（順序）asyncio的高級用法：
import asyncio
from typing import AsyncGenerator
# 異步生成器
asyncdefasync_generator():
for i in range(5):
await asyncio.sleep(1)
yield i
# 異步上下文管理器
classAsyncResource:
asyncdef__aenter__(self):
print("獲取資源")
await asyncio.sleep(1)
return self
asyncdef__aexit__(self, exc_type, exc_val, exc_tb):
print("釋放資源")
await asyncio.sleep(1)
# 使用
asyncdefuse_resource():
asyncwith AsyncResource() as resource:
print("使用資源")
asyncio.run(use_resource())asyncio的性能特性：
import asyncio
import time
import aiohttp
asyncdefbenchmark_asyncio():
"""測試asyncio處理1000個并發請求的性能"""
start = time.time()
asyncdeffetch(session, url):
try:
asyncwith session.get(url, timeout=5) as response:
returnawait response.text()
except:
returnNone
asyncwith aiohttp.ClientSession() as session:
tasks = [fetch(session, f"http://example.com/{i}") for i in range(1000)]
results = await asyncio.gather(*tasks)
end = time.time()
print(f"asyncio耗時：{end - start:.2f}秒，成功請求：{len([r for r in results if r])}")
asyncio.run(benchmark_asyncio())方案二：threading - 輕量級并發
threading的適用場景：
import threading
import time
from queue import Queue
defworker(queue, worker_id):
"""工作線程"""
whileTrue:
task = queue.get()
if task isNone:
break
print(f"Worker {worker_id} 處理任務 {task}")
time.sleep(1)
queue.task_done()
defmain_threading():
queue = Queue()
num_workers = 4
# 創建并啟動工作線程
threads = []
for i in range(num_workers):
t = threading.Thread(target=worker, args=(queue, i))
t.start()
threads.append(t)
# 添加任務
for i in range(10):
queue.put(i)
# 等待所有任務完成
queue.join()
# 停止工作線程
for _ in range(num_workers):
queue.put(None)
for t in threads:
t.join()
main_threading()threading的局限性：
import threading
import time
defcpu_intensive():
"""CPU密集型計算"""
total = 0
for i in range(100000000):
total += i
return total
# 單線程執行
start = time.time()
cpu_intensive()
cpu_intensive()
print(f"單線程耗時：{time.time() - start:.2f}秒")
# 多線程執行（受GIL影響，實際更慢）
start = time.time()
t1 = threading.Thread(target=cpu_intensive)
t2 = threading.Thread(target=cpu_intensive)
t1.start()
t2.start()
t1.join()
t2.join()
print(f"多線程耗時：{time.time() - start:.2f}秒")
# 結果：多線程因GIL競爭反而更慢！threading的正確用途：
import threading
import time
import requests
from concurrent.futures import ThreadPoolExecutor
deffetch_url(url):
"""獲取URL內容"""
try:
response = requests.get(url, timeout=5)
return len(response.content)
except:
return0
defbenchmark_threading():
urls = ["http://example.com"] * 100
# 使用線程池
with ThreadPoolExecutor(max_workers=10) as executor:
sizes = list(executor.map(fetch_url, urls))
print(f"成功獲取 {len([s for s in sizes if s > 0])} 個URL")
benchmark_threading()方案三：multiprocessing - CPU密集型的利器
multiprocessing基礎：
import multiprocessing
import time
defcpu_intensive(n):
"""CPU密集型計算"""
total = 0
for i in range(n):
total += i ** 2
return total
defmain_multiprocessing():
# 創建進程池
with multiprocessing.Pool(processes=4) as pool:
tasks = [100000000] * 4
start = time.time()
results = pool.map(cpu_intensive, tasks)
end = time.time()
print(f"多進程耗時：{end - start:.2f}秒")
print(f"結果：{results}")
if __name__ == '__main__':
main_multiprocessing()multiprocessing的進程間通信：
import multiprocessing
from multiprocessing import Queue, Pipe
defworker_queue(queue):
"""通過隊列通信"""
queue.put("Message from worker")
defmain():
# 方法1：使用隊列
queue = multiprocessing.Queue()
p = multiprocessing.Process(target=worker_queue, args=(queue,))
p.start()
message = queue.get()
print(f"收到消息：{message}")
p.join()
# 方法2：使用管道
parent_conn, child_conn = multiprocessing.Pipe()
defworker_pipe(conn):
conn.send("Hello from pipe")
conn.close()
p = multiprocessing.Process(target=worker_pipe, args=(child_conn,))
p.start()
message = parent_conn.recv()
print(f"收到消息：{message}")
p.join()
if __name__ == '__main__':
main()性能對比與最佳實踐
綜合性能測試：
import asyncio
import threading
import multiprocessing
import time
import requests
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
defbenchmark_all():
"""綜合性能對比"""
# 測試1：I/O密集型（網絡請求）
defio_task():
try:
requests.get("http://httpbin.org/delay/1", timeout=5)
return1
except:
return0
# asyncio版本
asyncdefio_asyncio():
import aiohttp
asyncwith aiohttp.ClientSession() as session:
tasks = []
for _ in range(10):
tasks.append(io_asyncio_task(session))
returnawait asyncio.gather(*tasks)
asyncdefio_asyncio_task(session):
try:
asyncwith session.get("http://httpbin.org/delay/1", timeout=5) as r:
return1
except:
return0
# threading版本
defio_threading():
with ThreadPoolExecutor(max_workers=10) as executor:
return list(executor.map(lambda _: io_task(), range(10)))
# 測試CPU密集型
defcpu_task():
return sum(i ** 2for i in range(10000000))
# threading版本（CPU密集型）
defcpu_threading():
with ThreadPoolExecutor(max_workers=4) as executor:
return list(executor.map(lambda _: cpu_task(), range(4)))
# multiprocessing版本
defcpu_multiprocessing():
with ProcessPoolExecutor(max_workers=4) as executor:
return list(executor.map(lambda _: cpu_task(), range(4)))
print("=== 性能對比 ===")
# I/O測試
print("\n1. I/O密集型（10個網絡請求）")
start = time.time()
io_threading()
print(f"threading耗時：{time.time() - start:.2f}秒")
# CPU測試
print("\n2. CPU密集型（4個大計算）")
start = time.time()
cpu_threading()
print(f"threading耗時：{time.time() - start:.2f}秒")
start = time.time()
cpu_multiprocessing()
print(f"multiprocessing耗時：{time.time() - start:.2f}秒")
benchmark_all()選擇決策樹
┌─ 是I/O密集型嗎？
│ ├─ Yes ──> 是否需要實時響應？
│ │ ├─ Yes ──> 使用 asyncio（推薦）
│ │ └─ No ──> 可用threading或asyncio
│ │
│ └─ No ──> 是CPU密集型嗎？
│ ├─ Yes ──> 使用 multiprocessing
│ └─ No ──> 數據處理量??？
│ ├─ Yes ──> 單線程即可
│ └─ No ──> 使用threading實戰案例：混合方案
import asyncio
import multiprocessing
from concurrent.futures import ProcessPoolExecutor
asyncdefhybrid_approach():
"""混合使用異步和多進程"""
defcpu_intensive(n):
# CPU密集計算
return sum(i ** 2for i in range(n))
# 先用asyncio并發發起任務
loop = asyncio.get_event_loop()
with ProcessPoolExecutor(max_workers=4) as executor:
tasks = []
for i in range(10):
# 在線程池中運行CPU密集操作
task = loop.run_in_executor(executor, cpu_intensive, 10000000)
tasks.append(task)
results = await asyncio.gather(*tasks)
return results
asyncio.run(hybrid_approach())結尾
在Python并發編程中，沒有絕對的最優方案，只有最適合當前場景的方案。asyncio適合I/O密集型的高并發場景，threading提供了輕量級的并發支持，multiprocessing則是CPU密集型計算的終極武器。理解GIL的影響、掌握三種方案的特點和適用場景，選擇合適的方案進行組合使用，這才是實戰中的最佳實踐。隨著FastAPI等現代框架的普及，asyncio已經成為主流選擇，但在處理混合型應用時，仍然需要靈活地運用三種方案。
責任編輯：趙寧寧
來源：
Python數智工坊
