Async và await: lập trình bất đồng bộ trong Python

Bởi

Lập trình bất đồng bộ cho phép chương trình làm việc khác trong lúc chờ đợi. Nghe đơn giản, nhưng đây là chủ đề dễ hiểu sai nhất trong Python hiện đại.

Async giải quyết vấn đề gì?

Hãy hình dung bạn cần tải 100 trang web. Cách tuần tự: gửi request, chờ phản hồi, xử lý, rồi mới sang trang tiếp theo. Phần lớn thời gian CPU của bạn ngồi không, chỉ chờ mạng.

Async cho phép trong lúc chờ trang thứ nhất, chương trình gửi luôn request cho trang thứ hai, thứ ba, và cứ thế. Tổng thời gian giảm từ “tổng của mọi lần chờ” xuống còn “lần chờ lâu nhất”.

Điểm mấu chốt: async chỉ giúp với công việc chờ đợi vào ra — mạng, đĩa, cơ sở dữ liệu. Nó không giúp gì với công việc nặng CPU như xử lý ảnh hay tính toán khoa học. Với loại đó, bạn cần đa tiến trình.

Cú pháp cơ bản

import asyncio

async def chao(ten, cho):
    await asyncio.sleep(cho)
    print(f"Xin chào, {ten}")
    return ten

async def main():
    await chao("An", 1)
    await chao("Bình", 1)     # mất tổng 2 giây — vẫn tuần tự!

asyncio.run(main())

Đây là hiểu lầm phổ biến nhất: viết async và await không tự động làm mọi thứ chạy song song. await nghĩa là “dừng ở đây chờ xong”. Muốn chạy đồng thời, bạn phải nói rõ:

async def main():
    ket_qua = await asyncio.gather(
        chao("An", 1),
        chao("Bình", 1),
    )
    print(ket_qua)     # mất khoảng 1 giây

Task — chạy nền

async def main():
    tac_vu = asyncio.create_task(chao("An", 2))

    # Làm việc khác trong lúc chờ
    await chao("Bình", 1)

    ket_qua = await tac_vu     # lấy kết quả khi cần

Từ Python 3.11, cách được khuyến nghị là dùng nhóm tác vụ. Nó đảm bảo mọi tác vụ đều được dọn dẹp đúng cách khi có lỗi:

async def main():
    async with asyncio.TaskGroup() as nhom:
        t1 = nhom.create_task(chao("An", 2))
        t2 = nhom.create_task(chao("Bình", 1))
    # Ra khỏi khối này, cả hai chắc chắn đã xong
    print(t1.result(), t2.result())

Nếu một tác vụ trong nhóm ném ngoại lệ, các tác vụ còn lại sẽ bị hủy và ngoại lệ được ném ra ngoài. Cách cũ với gather dễ để lại tác vụ mồ côi chạy tiếp trong nền.

Ví dụ thực tế: tải nhiều trang

import asyncio
import httpx

async def tai_trang(client, url):
    try:
        r = await client.get(url, timeout=10)
        r.raise_for_status()
        return {"url": url, "do_dai": len(r.text)}
    except httpx.HTTPError as e:
        return {"url": url, "loi": str(e)}

async def tai_tat_ca(urls, dong_thoi=10):
    gioi_han = asyncio.Semaphore(dong_thoi)

    async def co_gioi_han(client, url):
        async with gioi_han:
            return await tai_trang(client, url)

    async with httpx.AsyncClient() as client:
        return await asyncio.gather(
            *(co_gioi_han(client, u) for u in urls)
        )

urls = [f"https://example.com/trang/{i}" for i in range(100)]
ket_qua = asyncio.run(tai_tat_ca(urls))

Semaphore giới hạn số request đồng thời. Không có nó, bạn bắn 100 kết nối cùng lúc — dễ bị máy chủ đối tác chặn, và cũng có thể làm cạn tài nguyên máy mình.

Sai lầm thường gặp

Gọi hàm đồng bộ chặn trong code async

async def sai():
    time.sleep(5)              # chặn TOÀN BỘ event loop 5 giây
    return "xong"

async def dung():
    await asyncio.sleep(5)     # nhường quyền cho tác vụ khác
    return "xong"

Cùng nguyên tắc này áp dụng cho mọi thư viện đồng bộ. Dùng requests trong hàm async sẽ chặn event loop; hãy dùng httpx hoặc aiohttp.

Khi buộc phải gọi code đồng bộ, đẩy nó sang luồng riêng:

ket_qua = await asyncio.to_thread(ham_dong_bo_cham, tham_so)

Quên await

async def main():
    chao("An", 1)      # tạo coroutine rồi vứt đi, không chạy gì cả
    await chao("An", 1)  # đúng

Python sẽ cảnh báo “coroutine was never awaited”, nhưng cảnh báo này dễ bị bỏ qua giữa hàng loạt dòng log khác.

Async với cơ sở dữ liệu

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")

async def lay_bai_viet(bai_id):
    async with AsyncSession(engine) as session:
        return await session.get(BaiViet, bai_id)

Chú ý driver phải là loại bất đồng bộ — asyncpg thay vì psycopg2. Dùng driver đồng bộ với engine async sẽ không hoạt động.

Khi nào KHÔNG dùng async

Async thêm độ phức tạp đáng kể: bạn phải chọn thư viện tương thích, phải hiểu event loop, và việc gỡ lỗi khó hơn. Đừng dùng khi:

  • Chương trình chủ yếu tính toán, không chờ vào ra
  • Chỉ có vài chục thao tác vào ra — chi phí phức tạp không đáng
  • Thư viện quan trọng bạn cần chưa hỗ trợ async
  • Script chạy một lần rồi thôi

Với công việc nặng CPU, dùng đa tiến trình:

from concurrent.futures import ProcessPoolExecutor

with ProcessPoolExecutor() as executor:
    ket_qua = list(executor.map(xu_ly_anh, danh_sach_anh))

Tóm lại

Ba câu đáng nhớ:

  • await nghĩa là “chờ ở đây”, không phải “chạy song song”
  • Muốn đồng thời, dùng asyncio.gather hoặc TaskGroup
  • Một lệnh gọi đồng bộ chặn ở bất kỳ đâu cũng làm hỏng toàn bộ lợi ích của async