asyncio hay threading? Chọn đúng trong 3 câu hỏi

1. Tổng quan

Nhiều người mới học Python concurrency hay hỏi “asyncio hay threading tốt hơn?” — câu hỏi này sai ngay từ đầu, vì hai công cụ giải quyết hai vấn đề khác nhau. Bài này đưa ra 3 câu hỏi cụ thể để bạn tự trả lời, kèm code minh họa chạy được. Đọc xong, bạn sẽ biết chính xác khi nào dùng threading, khi nào dùng asyncio, và khi nào cả hai đều vô dụng (phải dùng multiprocessing).

2. Vì sao câu hỏi “cái nào nhanh hơn” là sai

Python có Global Interpreter Lock (GIL) — chỉ một thread Python được thực thi bytecode tại một thời điểm, bất kể máy có bao nhiêu CPU core. threading và asyncio đều không phá được GIL. Cả hai chỉ giúp CPU không bị “nghẽn” khi chờ I/O (đọc file, gọi network, truy vấn DB) — chúng không giúp tính toán nhanh hơn.

import time

def cpu_bound_task():
    total = 0
    for i in range(10_000_000):
        total += i * i
    return total

start = time.perf_counter()
cpu_bound_task()
cpu_bound_task()
print(f"Sequential: {time.perf_counter() - start:.2f}s")

Sequential: 1.85s

Dù bạn chạy hai lần cpu_bound_task() bằng threading.Thread hay asyncio.gather, tổng thời gian không giảm, vì GIL chỉ cho một thread chạy Python code tại một thời điểm. Đây là lý do câu hỏi đầu tiên dưới đây quan trọng nhất.

Lưu ý: Nếu bài toán là CPU-bound (tính toán số học nặng, xử lý ảnh, mã hóa…), cả threading và asyncio đều không giúp gì. Bạn cần multiprocessing hoặc viết phần nặng bằng C/Rust extension (NumPy, Cython).

3. Câu hỏi 1: Công việc của bạn I/O-bound hay CPU-bound?

Đây là câu hỏi quyết định bạn có nên dùng concurrency kiểu “chờ” (I/O-bound) hay không.

3.1. I/O-bound: cả threading và asyncio đều hiệu quả

import time
import requests
from concurrent.futures import ThreadPoolExecutor

URLS = ["https://httpbin.org/delay/1"] * 5

def fetch(url: str) -> int:
    return requests.get(url).status_code

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5) as pool:
    results = list(pool.map(fetch, URLS))
print(f"Threading: {time.perf_counter() - start:.2f}s, results={results}")

Threading: 1.12s, results=[200, 200, 200, 200, 200]

ThreadPoolExecutor chạy 5 request đồng thời bằng 5 OS thread. Vì requests.get là blocking call chờ network, GIL được nhả ra khi thread đang chờ I/O ở tầng C — nên 5 request chạy song song thật, tổng thời gian ~1s thay vì 5s.

import asyncio
import aiohttp
import time

URLS = ["https://httpbin.org/delay/1"] * 5

async def fetch(session: aiohttp.ClientSession, url: str) -> int:
    async with session.get(url) as resp:
        return resp.status

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in URLS]
        return await asyncio.gather(*tasks)

start = time.perf_counter()
results = asyncio.run(main())
print(f"Asyncio: {time.perf_counter() - start:.2f}s, results={results}")

Asyncio: 1.08s, results=[200, 200, 200, 200, 200]

Kết quả tương đương threading, nhưng cơ chế khác: chỉ một thread duy nhất, event loop chủ động await và nhường quyền điều khiển khi chờ I/O, không cần OS tạo thread mới.

3.2. CPU-bound: cả hai đều thua multiprocessing

from concurrent.futures import ProcessPoolExecutor

def cpu_bound_task():
    return sum(i * i for i in range(10_000_000))

start = time.perf_counter()
with ProcessPoolExecutor(max_workers=2) as pool:
    results = list(pool.map(cpu_bound_task, [None, None]))
print(f"Multiprocessing: {time.perf_counter() - start:.2f}s")

Multiprocessing: 1.02s

ProcessPoolExecutor tạo process riêng, mỗi process có GIL riêng, nên tận dụng được nhiều CPU core thật. Nếu bạn đang phân vân “threading hay asyncio” cho bài toán CPU-bound, câu trả lời đúng là cả hai đều sai — hãy dùng multiprocessing.

4. Câu hỏi 2: Hệ sinh thái bạn dùng có hỗ trợ async không?

Đây là câu hỏi thực dụng nhất khi làm việc trong codebase thật, nơi bạn không viết mọi thứ từ đầu.

import psycopg2  # driver đồng bộ, blocking

def get_user(user_id: int) -> dict:
    conn = psycopg2.connect("dbname=app")
    cur = conn.cursor()
    cur.execute("SELECT id, name FROM users WHERE id = %s", (user_id,))
    row = cur.fetchone()
    conn.close()
    return {"id": row[0], "name": row[1]}

psycopg2 là thư viện blocking — nó không có phiên bản await. Muốn chạy song song nhiều query với driver này, threading (hoặc run_in_executor của asyncio) là lựa chọn tự nhiên, vì bạn không cần viết lại gì cả.

import asyncpg  # driver async-native

async def get_user(conn: asyncpg.Connection, user_id: int) -> dict:
    row = await conn.fetchrow(
        "SELECT id, name FROM users WHERE id = $1", user_id
    )
    return {"id": row["id"], "name": row["name"]}

Nếu hệ sinh thái của bạn đã có driver async-native (asyncpg, aiohttp, motor cho MongoDB…), asyncio phát huy tối đa lợi thế: hàng nghìn connection đồng thời trên một thread, không tốn chi phí tạo OS thread.

Lưu ý: Trộn code blocking vào trong async def mà không await đúng cách (ví dụ gọi trực tiếp requests.get trong coroutine) sẽ block toàn bộ event loop — mọi task khác phải chờ, kể cả những task không liên quan. Đây là lỗi phổ biến nhất khi mới chuyển từ threading sang asyncio.

5. Câu hỏi 3: Bạn cần bao nhiêu task đồng thời?

Chi phí tài nguyên giữa thread và coroutine chênh lệch rất lớn khi số lượng task tăng.

import threading
import resource

def noop():
    threading.Event().wait(2)

threads = [threading.Thread(target=noop) for _ in range(2000)]
for t in threads:
    t.start()

peak_kb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
print(f"Peak RSS với 2000 threads: {peak_kb / 1024:.1f} MB")

for t in threads:
    t.join()

Peak RSS với 2000 threads: 1850.3 MB

Mỗi OS thread mặc định chiếm vài trăm KB đến vài MB stack, nên 2000 thread dễ dàng ngốn gần 2 GB RAM và tạo áp lực context-switching lớn cho OS scheduler.

import asyncio
import resource

async def noop():
    await asyncio.sleep(2)

async def main():
    tasks = [asyncio.create_task(noop()) for _ in range(2000)]
    await asyncio.gather(*tasks)
    peak_kb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
    print(f"Peak RSS với 2000 coroutines: {peak_kb / 1024:.1f} MB")

asyncio.run(main())

Peak RSS với 2000 coroutines: 95.7 MB

Coroutine là object Python thuần, không cần OS thread riêng — chi phí tạo 2000 task chỉ bằng vài chục MB. Nếu bạn cần xử lý hàng nghìn kết nối đồng thời (web server, chat server, crawler quy mô lớn), asyncio là lựa chọn gần như bắt buộc; threading sẽ hết tài nguyên trước khi hết băng thông.

6. Bảng quyết định nhanh

Câu hỏi	Trả lời → Lựa chọn
1. I/O-bound hay CPU-bound?	CPU-bound → multiprocessing. I/O-bound → tiếp câu 2
2. Thư viện bạn dùng có async-native không?	Không (driver blocking cũ) → threading. Có (aiohttp, asyncpg…) → asyncio
3. Cần bao nhiêu task đồng thời?	Vài chục → threading đủ đơn giản. Hàng trăm/nghìn → asyncio

Trong thực tế, nhiều hệ thống dùng cả hai: asyncio cho tầng network I/O-bound số lượng lớn, kết hợp run_in_executor chạy các đoạn code blocking (ví dụ thư viện cũ) trong thread pool riêng để không chặn event loop.

import asyncio

def legacy_blocking_call(x: int) -> int:
    import time
    time.sleep(1)  # giả lập thư viện blocking cũ
    return x * 2

async def main():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, legacy_blocking_call, 21)
    print(result)

asyncio.run(main())

42

run_in_executor đẩy hàm blocking sang một thread trong thread pool mặc định, event loop vẫn tự do xử lý các coroutine khác trong lúc chờ.

7. Kiểm thử hành vi với pytest

Ví dụ test xác nhận asyncio.gather chạy các task song song thật, không phải tuần tự:

import asyncio
import time
import pytest

async def sleepy_task(seconds: float) -> float:
    await asyncio.sleep(seconds)
    return seconds

@pytest.mark.asyncio
async def test_gather_runs_concurrently():
    start = time.perf_counter()
    results = await asyncio.gather(
        sleepy_task(0.5), sleepy_task(0.5), sleepy_task(0.5)
    )
    elapsed = time.perf_counter() - start

    assert results == [0.5, 0.5, 0.5]
    # Nếu chạy tuần tự sẽ mất ~1.5s, chạy song song chỉ ~0.5s
    assert elapsed < 0.7

Dependency cần thiết cho test:

pytest
pytest-asyncio

Test này khẳng định đúng bản chất asyncio: ba task sleep(0.5) chạy đồng thời trên một thread chỉ tốn ~0.5s tổng, chứ không phải 1.5s — nếu assertion elapsed < 0.7 fail, nghĩa là code của bạn đang await tuần tự thay vì gather.

8. Lỗi hay gặp

• Dùng time.sleep() trong coroutine: phải dùng await asyncio.sleep(). time.sleep() là blocking call, chặn đứng toàn bộ event loop.
• Quên await: gọi fetch(session, url) mà không await chỉ tạo ra coroutine object, không chạy gì cả — Python sẽ cảnh báo RuntimeWarning: coroutine was never awaited.
• Chia sẻ state giữa threads mà không lock: threading có race condition thật (nhiều thread cùng ghi một biến), cần threading.Lock. asyncio trên một thread duy nhất không có race condition kiểu này, nhưng vẫn có thể có bug logic nếu await ở giữa đoạn code tưởng là “atomic”.
• Dùng ThreadPoolExecutor cho bài toán CPU-bound rồi thất vọng vì không nhanh hơn — vì GIL vẫn chặn, hãy dùng ProcessPoolExecutor.

Lưu ý: Nếu team của bạn đã quen threading và codebase toàn thư viện blocking, đừng vội refactor sang asyncio chỉ vì nó “mới hơn”. Chi phí viết lại toàn bộ driver, xử lý exception khác đi, và debug stack trace coroutine phức tạp hơn nhiều so với lợi ích nhận được nếu số lượng task đồng thời không lớn.

9. Kết luận

Chọn giữa asyncio và threading không phải chuyện “cái nào nhanh hơn” mà là ba câu hỏi cụ thể: bài toán I/O-bound hay CPU-bound, hệ sinh thái thư viện có async-native không, và quy mô đồng thời bạn cần xử lý. CPU-bound luôn cần multiprocessing. I/O-bound với thư viện cũ và số lượng task nhỏ thì threading đơn giản, dễ tích hợp. I/O-bound quy mô lớn với thư viện async-native thì asyncio là lựa chọn đúng về hiệu năng và tài nguyên.