Async Python: asyncio, async/await

1. Concurrency vs Parallelism

Hai khái niệm này thường bị nhầm lẫn vì cùng gợi ý "làm nhiều thứ một lúc". Nhưng ở mức kỹ thuật, chúng không giống nhau:

Một chương trình có thể là (concurrent + không parallel) — ví dụ asyncio chạy 1000 coroutine trên 1 thread. Có thể là (parallel + không concurrent) — ví dụ 4 tiến trình hoàn toàn độc lập, không phối hợp gì. Có thể vừa concurrent vừa parallel — web server thực tế: nhiều worker process, mỗi process chạy event loop với nhiều coroutine.

2. GIL — Global Interpreter Lock

GIL là một mutex toàn cục trong CPython (bản tham chiếu của Python). Quy tắc: tại bất kỳ thời điểm nào, chỉ duy nhất một thread được phép thực thi Python bytecode. Các thread khác phải chờ.

Hệ quả nghe có vẻ "tiêu cực" nhưng cần hiểu chính xác:

• Threading không giúp tăng tốc CPU-bound trong Python thuần. Hai thread cùng tính toán pi — vẫn chỉ một CPU làm việc.
• Nhưng threading vẫn lợi cho I/O-bound: khi một thread block I/O (đọc socket, đọc file), nó giải phóng GIL cho thread khác chạy.
• Các thư viện C (NumPy, SciPy, Pillow) thường release GIL trong vòng tính toán C, nên threading có thể song song hoá những phần đó.
• Để parallel CPU-bound trong Python thuần, dùng multiprocessing — mỗi process có GIL riêng.
• Python 3.13 đã có chế độ thử nghiệm "no-GIL" (PEP 703 / free-threaded build), nhưng chưa phổ biến trong production năm 2026.

3. So sánh 4 mô hình

Bảng so sánh tổng quan giúp chọn đúng công cụ cho đúng workload:

4. Coroutine — đơn vị async cơ bản

Trong Python, coroutine là một function được khai báo bằng async def. Khi gọi coroutine function, bạn không chạy nó — bạn nhận được một coroutine object sẵn sàng được schedule.

import asyncio

async def fetch(url):
    # giả lập I/O
    await asyncio.sleep(1)
    return f"data from {url}"

# Gọi fetch(...) KHÔNG chạy code bên trong:
coro = fetch("https://example.com")
print(coro)
# <coroutine object fetch at 0x...>
# Phải schedule bằng await, asyncio.run, hoặc create_task

result = asyncio.run(coro)
print(result)  # 'data from https://example.com'

async def work():
    print("đang làm")

async def main():
    work()           # SAI — chỉ tạo coro, không chạy
    await work()     # ĐÚNG — schedule và đợi

5. await — tạm dừng và nhường quyền

await <awaitable> là biểu thức làm 3 việc:

1. Tạm dừng coroutine hiện tại.
2. Nhường quyền điều khiển về event loop để task khác có cơ hội chạy.
3. Khi awaitable hoàn thành, tiếp tục coroutine từ chỗ tạm dừng với giá trị trả về.

await chỉ được phép dùng bên trong async def. Trong sync function thông thường sẽ là SyntaxError.

import asyncio, time

async def say(name, delay):
    print(f"{name} bắt đầu")
    await asyncio.sleep(delay)
    print(f"{name} xong")
    return name

async def main():
    t0 = time.perf_counter()
    a = await say("A", 1)   # dừng 1s, sau đó chạy tiếp
    b = await say("B", 2)   # dừng 2s nữa
    print(f"Tổng: {time.perf_counter()-t0:.1f}s")

asyncio.run(main())
"""
A bắt đầu
A xong
B bắt đầu
B xong
Tổng: 3.0s   # tuần tự
"""

Ở ví dụ trên, A và B chạy tuần tự vì mỗi await chờ xong mới chuyển sang dòng tiếp theo. Muốn song song phải dùng gather hoặc create_task — sẽ trình bày bên dưới.

6. asyncio.run — entry point

asyncio.run(coro) là cách khuyến nghị để khởi động ứng dụng asyncio ở top-level (script, file __main__). Nó:

• Tạo một event loop mới.
• Chạy coro cho đến khi xong.
• Đóng tất cả async generator, shutdown executor, đóng loop.
• Trả về kết quả của coro.

import asyncio

async def main():
    # Toàn bộ code async sống ở đây
    print("hello async")

if __name__ == "__main__":
    asyncio.run(main())

async def main():
    asyncio.run(other())   # SAI — RuntimeError

Trong các framework tích hợp sẵn loop (FastAPI, aiohttp.web, Jupyter, Django async views), bạn không gọi asyncio.run — framework đã quản lý loop. Chỉ dùng nó cho script/CLI tự viết.

7. asyncio.gather — chạy song song

asyncio.gather(*aws) nhận nhiều awaitable và schedule chúng cùng lúc, trả về một awaitable mới. Khi await nó, bạn nhận được danh sách kết quả theo đúng thứ tự đầu vào.

import asyncio, time

async def fetch(name, delay):
    await asyncio.sleep(delay)
    return f"<{name}>"

async def main():
    t0 = time.perf_counter()
    results = await asyncio.gather(
        fetch("A", 1),
        fetch("B", 2),
        fetch("C", 3),
    )
    print(results, f"{time.perf_counter()-t0:.1f}s")
    # ['<A>', '<B>', '<C>'] 3.0s  (chứ không phải 6.0s)

asyncio.run(main())

Lưu ý: gather nhận awaitable trực tiếp làm tham số, không nhận list. gather([c1, c2]) sẽ không chạy mà raise lỗi. Muốn unpack list dùng dấu sao: gather(*[c1, c2]).

urls = ["u1", "u2", "u3"]
coros = [fetch(u, 1) for u in urls]
results = await asyncio.gather(*coros)

async def might_fail(x):
    if x < 0: raise ValueError("âm")
    return x * 2

results = await asyncio.gather(
    might_fail(1), might_fail(-1), might_fail(3),
    return_exceptions=True,
)
# [2, ValueError('âm'), 6]

8. asyncio.create_task — fire-and-forget

gather tốt khi bạn biết trước danh sách coroutine cần đợi. Khi muốn schedule một coroutine chạy nền mà không chờ ngay, dùng asyncio.create_task(coro):

• Trả về một Task object (subclass của Future).
• Task bắt đầu chạy ở lần await tiếp theo của coroutine hiện tại.
• Có thể await task sau để lấy kết quả, hoặc task.cancel() để huỷ.

import asyncio

async def heartbeat():
    while True:
        print("ping")
        await asyncio.sleep(1)

async def main():
    bg = asyncio.create_task(heartbeat())   # start nền
    try:
        await do_work()                          # làm việc chính
    finally:
        bg.cancel()                                # dọn task nền
        try:
            await bg
        except asyncio.CancelledError:
            pass

# SAI — Python có thể GC task chưa xong
asyncio.create_task(background_job())

# ĐÚNG — giữ tham chiếu
task = asyncio.create_task(background_job())
# hoặc lưu vào set:
_bg_tasks: set = set()
t = asyncio.create_task(background_job())
_bg_tasks.add(t)
t.add_done_callback(_bg_tasks.discard)

9. asyncio.wait_for — timeout

Cuộc sống async đầy rủi ro: một request HTTP có thể treo mãi, một database query có thể tắc. Để áp dụng timeout cho một coroutine, dùng asyncio.wait_for(aw, timeout):

import asyncio

async def slow():
    await asyncio.sleep(10)
    return "done"

async def main():
    try:
        result = await asyncio.wait_for(slow(), timeout=2)
    except asyncio.TimeoutError:
        print("quá hạn 2s")
        result = None
    return result

asyncio.run(main())

Khi hết thời gian, wait_for sẽ cancel coroutine bên trong rồi raise TimeoutError. Quan trọng: coroutine bên trong phải "co-operative" — nó cần đến điểm await để nhận tín hiệu cancel. Nếu nó đang block CPU (vòng for tính toán nặng) thì timeout không tới được.

async with asyncio.timeout(2):
    result = await slow()
# Pythonic hơn, có thể bao nhiều await

10. asyncio.Queue — Producer–Consumer

asyncio.Queue là FIFO queue an toàn cho coroutine, không cần lock (vì single-threaded). Đây là cách kinh điển để tách producer (đẩy việc) và consumer (xử lý):

import asyncio, random

async def producer(q: asyncio.Queue):
    for i in range(10):
        item = f"item-{i}"
        await q.put(item)
        print(f"đẩy {item}")
        await asyncio.sleep(random.uniform(0.1, 0.3))
    await q.put(None)  # sentinel báo hết

async def consumer(q: asyncio.Queue, name: str):
    while True:
        item = await q.get()
        if item is None:
            await q.put(None)  # forward cho consumer khác
            break
        print(f"{name} xử lý {item}")
        await asyncio.sleep(0.5)
        q.task_done()

async def main():
    q = asyncio.Queue(maxsize=5)
    await asyncio.gather(
        producer(q),
        consumer(q, "c1"),
        consumer(q, "c2"),
    )

asyncio.run(main())

API quan trọng của Queue:

• await q.put(item) — đẩy, block nếu đầy (khi có maxsize).
• await q.get() — lấy, block nếu rỗng.
• q.put_nowait(item), q.get_nowait() — bản non-blocking, raise nếu không thể.
• q.task_done() kết hợp với await q.join() — pattern "đợi tất cả item được xử lý xong".
• asyncio.PriorityQueue, asyncio.LifoQueue — biến thể.

11. async with và async for

Hai cú pháp hỗ trợ context manager và iterator có chứa code async:

11.1. async with — async context manager

Tương tự with, nhưng __enter__/__exit__ được thay bằng phiên bản async __aenter__/__aexit__. Phổ biến nhất: HTTP session, DB connection, file async.

import aiohttp

async def fetch_json(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()
# Session auto-close (close socket pool) khi rời block.

Tự định nghĩa async context manager với contextlib.asynccontextmanager:

from contextlib import asynccontextmanager

@asynccontextmanager
async def db_connection(dsn):
    conn = await connect(dsn)
    try:
        yield conn
    finally:
        await conn.close()

async with db_connection("postgres://...") as db:
    rows = await db.fetch("SELECT 1")

11.2. async for — async iterator

Iterator có __aiter__ trả về self, __anext__ là coroutine trả về phần tử kế (hoặc raise StopAsyncIteration). Một async generator (function vừa có async def vừa có yield) là cách dễ nhất để tạo:

import asyncio

async def stream_lines(file):
    while True:
        line = await file.readline()
        if not line: break
        yield line.decode().strip()

async def main():
    async for line in stream_lines(open_socket):
        print(line)

Async iterator thực sự toả sáng khi nguồn dữ liệu tới dần: HTTP streaming, websocket, Kafka consumer, DB cursor.

12. Event Loop của asyncio

Event loop của Python rất giống event loop của JS: single-threaded, chạy task đến khi gặp await, yield, sau đó pick task tiếp theo.

13. Sync vs Async — không trộn lung tung

Quy tắc vàng: code sync nặng KHÔNG được phép chạy trong coroutine. Vì asyncio chỉ có một thread, một dòng time.sleep(5) sẽ đóng băng toàn bộ event loop — mọi connection khác phải chờ.

import asyncio, time

async def bad():
    time.sleep(5)         # SAI — block toàn loop 5 giây

async def good():
    await asyncio.sleep(5) # ĐÚNG — yield, các task khác tự do chạy

Vấn đề là không phải lúc nào cũng có "phiên bản async" tương ứng. Khi buộc phải gọi function sync nặng (CPU hoặc I/O không có phiên bản async), dùng loop.run_in_executor để đẩy ra thread pool:

import asyncio
from concurrent.futures import ThreadPoolExecutor

def sync_cpu(n):
    # tính toán nặng, không có phiên bản async
    return sum(i * i for i in range(n))

async def main():
    loop = asyncio.get_running_loop()
    # None = dùng default executor (ThreadPoolExecutor)
    result = await loop.run_in_executor(None, sync_cpu, 10_000_000)
    print(result)

    # Python 3.9+: shortcut
    result = await asyncio.to_thread(sync_cpu, 10_000_000)

• ThreadPoolExecutor phù hợp khi sync function là I/O-bound (network legacy lib, disk).
• ProcessPoolExecutor phù hợp khi sync function là CPU-bound thuần Python (vì GIL).
• asyncio.to_thread(fn, *args) là wrapper gọn cho ThreadPool default — dùng từ Python 3.9.

14. Hệ sinh thái async

asyncio chỉ là core. Sức mạnh thực sự đến từ thư viện async-aware. Một số tên quan trọng năm 2026:

import asyncio, httpx

async def fetch(client, url):
    r = await client.get(url)
    return r.status_code, len(r.content)

async def main(urls):
    async with httpx.AsyncClient(timeout=5) as c:
        results = await asyncio.gather(
            *[fetch(c, u) for u in urls],
            return_exceptions=True,
        )
    return results

urls = [f"https://httpbin.org/delay/{i}" for i in range(10)]
asyncio.run(main(urls))

Bài tập

import asyncio, time, httpx

async def fetch(url, session):
    r = await session.get(url)
    return url, r.status_code, len(r.content)

async def fetch_sequential(urls):
    async with httpx.AsyncClient(timeout=5) as s:
        out = []
        for u in urls:
            out.append(await fetch(u, s))
        return out

async def fetch_parallel(urls):
    async with httpx.AsyncClient(timeout=5) as s:
        return await asyncio.gather(*[fetch(u, s) for u in urls])

async def main():
    urls = ["https://httpbin.org/delay/1"] * 10
    for name, fn in [("seq", fetch_sequential), ("par", fetch_parallel)]:
        t = time.perf_counter()
        await fn(urls)
        print(name, time.perf_counter() - t)
asyncio.run(main())

import asyncio, httpx

async def download_all(urls, concurrency=5):
    sem = asyncio.Semaphore(concurrency)
    async with httpx.AsyncClient(timeout=10) as s:
        async def one(u):
            async with sem:
                r = await s.get(u)
                return u, r.status_code

        return await asyncio.gather(*[one(u) for u in urls])

import asyncio

async def with_timeout(coro, seconds):
    try:
        return await asyncio.wait_for(coro, timeout=seconds)
    except asyncio.TimeoutError:
        return None

# So sánh:
# asyncio.wait_for RAISE TimeoutError → caller phải try/except.
# with_timeout swallows → caller chỉ check None.
#
# Trade-off: gọn cho caller nhưng mất thông tin (None vs giá trị bình
# thường có thể cũng là None). Cách an toàn hơn:

_SENTINEL = object()

async def with_timeout(coro, seconds, default=_SENTINEL):
    try:
        return await asyncio.wait_for(coro, seconds)
    except asyncio.TimeoutError:
        if default is _SENTINEL: raise
        return default

import asyncio, httpx

async def producer(q, urls):
    for u in urls:
        await q.put(u)
    for _ in range(5):
        await q.put(None)  # 1 sentinel/worker

async def worker(name, q, client):
    while True:
        url = await q.get()
        if url is None:
            break
        try:
            r = await client.get(url, timeout=5)
            print(name, url, r.status_code)
        except Exception as e:
            print(name, url, "ERR", e)

async def main(urls):
    q = asyncio.Queue(maxsize=20)
    async with httpx.AsyncClient() as client:
        workers = [
            asyncio.create_task(worker(f"w{i}", q, client))
            for i in range(5)
        ]
        await producer(q, urls)
        await asyncio.gather(*workers)

asyncio.run(main(["https://httpbin.org/get"] * 20))

import asyncio, time

async def slow_task(name):
    print(f"{name} start")
    time.sleep(2)
    print(f"{name} done")

async def main():
    await asyncio.gather(slow_task("A"), slow_task("B"), slow_task("C"))

asyncio.run(main())

async def slow_task(name):
    print(f"{name} start")
    await asyncio.sleep(2)
    print(f"{name} done")

async def slow_task(name):
    print(f"{name} start")
    await asyncio.to_thread(time.sleep, 2)
    print(f"{name} done")

Quiz

# A
await fetch_a()
await fetch_b()

# B
await asyncio.gather(fetch_a(), fetch_b())

Tổng kết

Sau chương 8, bạn nên đã master:

• Concurrency vs Parallelism: concurrency = interleave (1 CPU đủ), parallelism = chạy thật cùng lúc (nhiều CPU).
• GIL: rào cản threading CPU-bound trong CPython, nhưng I/O-bound vẫn dùng threading tốt.
• 4 mô hình: sync (đơn giản), threading (I/O nhẹ), multiprocessing (CPU thật), asyncio (I/O nặng).
• Coroutine: async def tạo coroutine; gọi không chạy, phải schedule.
• await: tạm dừng, yield cho event loop, tiếp tục khi awaitable xong. Chỉ trong async function.
• asyncio.run: entry-point top-level, không gọi từ trong coroutine.
• gather: chạy nhiều awaitable song song, trả list kết quả theo thứ tự.
• create_task: schedule nền, trả Task object. Nhớ giữ reference.
• wait_for / asyncio.timeout: áp dụng timeout, cancel nếu quá hạn.
• Queue: producer-consumer không cần lock vì single-threaded.
• async with / async for: context manager + iterator async — cho session, stream, cursor.
• Event loop asyncio: single-threaded, selector-based, giống JS event loop về tinh thần nhưng đơn giản hơn (1 queue).
• Sync vs async: KHÔNG block event loop. Sync nặng → run_in_executor / to_thread.
• Ecosystem: aiohttp, httpx, aiofiles, asyncpg, motor, redis.asyncio, FastAPI, anyio.

Kết nối

• Chương 6 (Generator) — async generator là sự kết hợp generator + coroutine; async for hoạt động trên async iterator.
• Chương 7 (Decorators) — decorator có thể wrap coroutine; nhớ check asyncio.iscoroutinefunction để xử lý đúng.
• Chương 9 (Typing) — type hint cho async: Awaitable[T], Coroutine[Any, Any, T], AsyncIterator[T].
• Chương 10 (Testing) — test async với pytest-asyncio: marker @pytest.mark.asyncio, fixture async.
• JavaScript Chương 6 (Async) — đối chiếu event loop JS với asyncio: cùng mental model, nhưng JS có micro/macro queue rõ ràng, JS auto-start loop còn Python cần asyncio.run.
• Dart Chương 7 (async/await) — Dart có cú pháp async/await rất giống Python; Future ~ Python coroutine; Stream ~ async iterator.
• Networking (Pillar 4) — asyncio dùng OS selector (epoll/kqueue/IOCP); hiểu TCP/HTTP giúp debug socket leak, connection pool.
• System Design (Pillar 7) — async I/O là nền tảng cho high-concurrency server (10K+ connection), pattern phổ biến trong microservice và real-time system.