Concurrency trong Python: So sánh các phương pháp cốt lõi
Nghẽn Network I/O gần như luôn là nguyên nhân chính khiến các web scraper, data pipeline và microservice bị đình trệ. Khi ứng dụng gửi một HTTP request hoặc truy vấn PostgreSQL, thread của hệ điều hành sẽ phải chờ từ 50 đến 300 mili-giây để nhận gói tin qua mạng.
Python cung cấp ba mô hình riêng biệt để giải phóng thời gian chờ vô ích này:
- Multi-threading (module
threading): Sử dụng các thread của hệ điều hành, chia sẻ chung bộ nhớ và nhường quyền thực thi khi gặp lời gọi blocking. Tuy nhiên, Global Interpreter Lock (GIL) của CPython giới hạn tính song song thực sự chỉ áp dụng được cho tác vụ I/O. Cơ chế context switching cũng gây ra chi phí phụ đáng kể cho hệ điều hành khi quy mô vượt qua vài trăm thread. - Multi-processing (module
multiprocessing): Khởi chạy các tiến trình Python độc lập, mỗi tiến trình sở hữu vùng nhớ và GIL riêng biệt. Cấu hình này cực kỳ lý tưởng cho các tác vụ nặng về CPU như nén ảnh hay nhân ma trận. Đáng tiếc là chi phí giao tiếp liên tiến trình (IPC) cùng mức chiếm dụng nền khoảng 30–50 MB cho mỗi worker khiến nó quá nặng nề khi phải xử lý hàng chục nghìn socket mạng mở đồng thời. - Cooperative Asynchronous I/O (
asyncio): Hoạt động bên trong một thread duy nhất thông qua cơ chế event loop. Các coroutine sẽ chủ động nhường lại quyền điều khiển cho loop bằng từ khóaawaitbất cứ khi nào chúng phải chờ I/O.
Ưu và nhược điểm: Thread vs Process vs AsyncIO
Lựa chọn sai mô hình sẽ dẫn đến các lỗi khó tái hiện, hiện tượng thread thrashing và lãng phí chi phí hạ tầng. Dưới đây là cách từng phương pháp hoạt động trong môi trường production:
1. Multi-Threading
- Ưu điểm: Tích hợp mượt mà với các thư viện đồng bộ cũ; bộ nhớ chia sẻ giúp loại bỏ chi phí tuần tự hóa (serialization).
- Nhược điểm: Hiện tượng race condition đòi hỏi phải sử dụng mutex lock cẩn thận; mỗi thread hệ điều hành chiếm mặc định 8 MB stack trên Linux; hiện tượng CPU cache churn làm sụt giảm thông lượng nghiêm trọng khi vượt quá 500 thread đồng thời.
2. Multi-Processing
- Ưu điểm: Vượt qua hoàn toàn rào cản GIL trên các CPU đa nhân; tính cô lập giữa các tiến trình đảm bảo sự cố sập ở một worker không bao giờ làm ảnh hưởng đến các tiến trình còn lại.
- Nhược điểm: Tốn nhiều bộ nhớ do phải nhân bản trạng thái interpreter; việc pickle các payload lớn qua ranh giới tiến trình tạo ra độ trễ serialization rõ rệt.
3. Python AsyncIO
- Ưu điểm: Cực kỳ nhẹ. Một coroutine chỉ tiêu tốn chưa đến 1 KB bộ nhớ, cho phép bạn quản lý hơn 20.000 kết nối mở trên một VPS khiêm tốn chỉ 2 GB RAM. Việc thực thi theo cơ chế xác định (deterministic) cũng loại bỏ hoàn toàn các lỗi race condition thường gặp khi lập lịch thread.
- Nhược điểm: Bất kỳ lời gọi blocking nào không được await—chẳng hạn như
time.sleep()hayrequests.get()—sẽ làm tắc nghẽn toàn bộ event loop. Ngoài ra, việc debug các chuỗi task lồng nhau cũng đòi hỏi bạn phải nắm vững cơ chế lập lịch coroutine.
Kiến trúc và thiết lập khuyến nghị
Đối với API gateway, scraper và event consumer, mô hình cooperative multitasking trên các phiên bản Python hiện đại (3.11+) đem lại thông lượng tối ưu nhất trên từng gigabyte RAM. Python 3.11 đã tăng tốc độ khởi tạo loop nội bộ thêm khoảng 10–15% và giới thiệu asyncio.TaskGroup, một context manager an toàn trước ngoại lệ giúp ngăn chặn tình trạng thất thoát hoặc mồ côi các background task.
Hãy giữ cho các dependency luôn gọn gàng. Trước khi thử nghiệm các ví dụ, hãy tạo một môi trường ảo và cài đặt một async HTTP client:
# Tạo và kích hoạt môi trường ảo riêng biệt
python3 -m venv .venv
source .venv/bin/activate
# Cài đặt HTTP client hiệu năng cao
pip install aiohttpHướng dẫn triển khai từng bước
Vấn đề: Xử lý đồng thời không kiểm soát làm sập upstream service
Hãy tưởng tượng bạn cần ping 500 microservice để lấy thông số health check. Chạy tuần tự sẽ mất hàng phút. Ngược lại, nếu gom cả 500 lời gọi vào một hàm asyncio.gather() đơn thuần, card mạng của bạn sẽ bị nghẽn vì 500 lượt bắt tay TCP đồng thời. Bạn sẽ sớm gặp phải tình trạng cạn kiệt socket, rớt kết nối và dính lỗi giới hạn tần suất HTTP 429.
Nguyên nhân gốc rễ: Giới hạn File Descriptor và quá tải Event Loop
Mỗi socket đang mở đều tiêu tốn một file descriptor. Trên Linux, các tiến trình thông thường có giới hạn mặc định là 1.024 descriptor mở (ulimit -n). Các coroutine không giới hạn sẽ cố gắng mở toàn bộ socket cùng lúc, chạm trần giới hạn của hệ điều hành và kích hoạt cơ chế phòng vệ của upstream service. Bạn cần có một người gác cổng.
Giải pháp thực tế: Kết hợp Event Loop, Task và Semaphore
Ba thành phần nguyên thủy sau đây sẽ tạo nên một async pipeline vững chắc:
- Event Loop: Bộ ghép kênh (multiplexer) sử dụng
epollhoặckqueueở tầng dưới để đánh thức coroutine khi dữ liệu từ socket truyền tới. - Task: Vỏ bọc dùng để lập lịch thực thi đồng thời cho coroutine trên event loop.
- Semaphore: Bộ đếm bất đồng bộ giới hạn số lượng coroutine được truy cập vào tài nguyên dùng chung cùng một lúc.
Dưới đây là đoạn mã hoàn chỉnh, sẵn sàng cho môi trường production để kiểm soát mức độ đồng thời:
import asyncio
import time
import aiohttp
# Giới hạn số lượng kết nối HTTP ra ngoài đồng thời
CONCURRENCY_LIMIT = 5
async def fetch_url(session: aiohttp.ClientSession, url: str, semaphore: asyncio.Semaphore) -> dict:
"""Lấy dữ liệu từ một endpoint duy nhất và tuân thủ giới hạn đồng thời."""
async with semaphore:
# Tối đa CONCURRENCY_LIMIT coroutine có thể vào khối này cùng lúc
start_time = time.monotonic()
try:
timeout = aiohttp.ClientTimeout(total=5)
async with session.get(url, timeout=timeout) as response:
data = await response.text()
elapsed = time.monotonic() - start_time
return {"url": url, "status": response.status, "duration": round(elapsed, 2), "size": len(data)}
except Exception as exc:
return {"url": url, "error": str(exc)}
async def main():
urls = [
f"https://httpbin.org/delay/{i % 2}"
for i in range(15)
]
semaphore = asyncio.Semaphore(CONCURRENCY_LIMIT)
# Sử dụng pooled TCP connector
connector = aiohttp.TCPConnector(limit=50)
async with aiohttp.ClientSession(connector=connector) as session:
# TaskGroup đảm bảo hủy bỏ sạch sẽ nếu có bất kỳ task nào gặp ngoại lệ chưa được xử lý
async with asyncio.TaskGroup() as tg:
tasks = [
tg.create_task(fetch_url(session, url, semaphore))
for url in urls
]
# Tất cả coroutine đều đã hoàn tất khi TaskGroup kết thúc
results = [task.result() for task in tasks]
successful = sum(1 for r in results if "status" in r and r["status"] == 200)
print(f"Đã xử lý xong {len(results)} request. Thành công: {successful}")
for item in results[:3]:
print(f" Mẫu: {item}")
if __name__ == "__main__":
# asyncio.run quản lý vòng đời của loop, signal handler và dọn dẹp thread
asyncio.run(main())
Xử lý mã Blocking bên trong AsyncIO
Các tác vụ nặng về CPU như tính mã băm HMAC, parse các payload JSON khổng lồ hay xử lý hình ảnh sẽ làm nghẽn event loop vốn chỉ chạy trên một thread. Trong lúc phép tính đó đang chạy, sẽ không có gói tin mạng nào được đọc hay gửi đi.
Để tránh tăng đột biến độ trễ, hãy ủy quyền các phép tính đồng bộ này cho một thread pool bằng asyncio.to_thread():
import hashlib
import asyncio
def heavy_hashing(payload: str) -> str:
"""Phép tính đồng bộ tiêu tốn nhiều tài nguyên CPU."""
result = payload
for _ in range(500_000):
result = hashlib.sha256(result.encode()).hexdigest()
return result
async def process_item(item: str):
# Chạy heavy_hashing trong ThreadPoolExecutor mặc định của Python mà không làm nghẽn loop
digest = await asyncio.to_thread(heavy_hashing, item)
print(f"Đã tính xong hash: {digest[:10]}...")
Việc kết hợp asyncio.Semaphore cho các thao tác mạng với asyncio.to_thread() cho các tác vụ tính toán sẽ giúp backend của bạn duy trì độ trễ cực kỳ ổn định và giới hạn tài nguyên một cách dễ đoán ngay cả dưới tải lưu lượng lớn.
