# Asyncio trong Python: async/await, TaskGroup, luồng, multiprocessing và GIL

URL: https://skillaria.top/vi/guides/python/asyncio-python-bat-dong-bo

Asyncio là thư viện bất đồng bộ tích hợp của Python: hàm khai báo bằng async def (coroutine) tạm dừng tại await, còn event loop chạy các tác vụ khác trong lúc đó, nên hàng trăm request mạng có thể diễn ra đồng thời trên một luồng. Với chờ mạng và đĩa, dùng asyncio hoặc threading; với tính toán nặng, dùng process (multiprocessing), vì GIL ngăn các luồng của CPython thông thường tính toán song song.

Asyncio là thư viện bất đồng bộ tích hợp sẵn của Python. Các hàm khai báo bằng `async def` có thể tạm dừng tại `await`, và trong lúc một hàm chờ phản hồi mạng, event loop chạy các tác vụ khác. Nhờ vậy hàng trăm request có thể đi cùng lúc trong một luồng. Nhưng asyncio chỉ là một trong ba công cụ: còn có luồng (`threading`) và tiến trình (`multiprocessing`). Dưới đây là cách asyncio python hoạt động, nên chọn công cụ nào cho từng bài toán, và GIL có liên quan gì ở đây.

## Chờ và tính toán là hai việc khác nhau

Không hiểu điểm này thì không thể nắm phần sau. Chương trình chậm vì một trong hai lý do:

- **I/O-bound** — đang chờ nhập/xuất: phản hồi API, cơ sở dữ liệu, đọc đĩa. CPU lúc này nhàn rỗi.
- **CPU-bound** — đang tính toán: xử lý ảnh, lặp qua hàng triệu bản ghi. CPU bận rộn.

Để chờ, asyncio hoặc luồng là phù hợp: trong khi một thao tác đang chờ, thao tác khác vẫn chạy. Để tính toán, cần song song thực sự trên nhiều nhân — dùng process. Nhầm lẫn giữa hai loại là lỗi lớn nhất: luồng không làm tính toán thuần Python nhanh hơn, còn dùng process cho ba request HTTP thì giống như dùng súng để đập muỗi.

## Asyncio là gì: coroutine và event loop

Code bất đồng bộ dựa trên hai từ khóa. `async def` khai báo một coroutine — hàm có thể tạm dừng. Khi gọi nó, thân hàm chưa chạy; thay vào đó tạo ra một đối tượng coroutine. `await` bên trong nói: «ở đây tôi sẽ chờ, còn bạn cứ làm việc khác». Mọi thứ được điều phối bởi event loop, khởi động qua `asyncio.run()`.

```
import asyncio

async def fetch_rate(currency):
    await asyncio.sleep(1)          # giả lập một request tới API
    return {"USD": 25450, "EUR": 27900, "CNY": 3550}[currency]

async def main():
    rate = await fetch_rate("USD")
    print(rate)

asyncio.run(main())  # 25450
```

`asyncio.sleep()` là một khoảng dừng bất đồng bộ: nó trả quyền điều khiển về cho event loop. Còn `time.sleep()` bên trong coroutine sẽ chặn toàn bộ event loop, và mọi tác vụ khác phải đứng chờ.

### Tuần tự hay đồng thời

Nếu viết `await` trong vòng lặp, các request sẽ chạy lần lượt — ba loại tiền tệ mất ba giây. Muốn chúng chạy cùng lúc, cần khởi chạy chúng dưới dạng task.

```
import asyncio, time

async def main():
    start = time.perf_counter()
    for cur in ("USD", "EUR", "CNY"):
        await fetch_rate(cur)                       # lần lượt
    print(f"{time.perf_counter() - start:.1f} s")   # 3.0 s

asyncio.run(main())
```

## TaskGroup: chạy cùng nhau và đợi tất cả

Cách hiện đại (Python 3.11 trở lên) là nhóm task. Trong khối `async with`, mỗi task khởi động ngay qua `tg.create_task()`, và khi thoát khỏi khối thì tất cả đã hoàn thành.

```
async def main():
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tasks = {cur: tg.create_task(fetch_rate(cur)) for cur in ("USD", "EUR", "CNY")}
    rates = {cur: t.result() for cur, t in tasks.items()}
    print(rates, f"{time.perf_counter() - start:.1f} s")   # ... 1.0 s

asyncio.run(main())
```

TaskGroup hoạt động nghiêm ngặt: nếu một task lỗi, các task còn lại bị hủy, còn mọi lỗi được gom vào `ExceptionGroup`. Bắt chúng bằng `except*`:

```
async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch_rate("USD"))
            tg.create_task(fetch_rate("XXX"))   # KeyError
    except* KeyError as eg:
        print("tiền tệ không xác định:", eg.exceptions)
```

### asyncio.gather và create_task

Trước khi có TaskGroup, `asyncio.gather()` là chuẩn: nó chạy các coroutine cùng lúc và trả về kết quả theo đúng thứ tự. Trong code cũ vẫn gặp nhiều.

```
async def main():
    usd, eur, cny = await asyncio.gather(
        fetch_rate("USD"), fetch_rate("EUR"), fetch_rate("CNY")
    )
    # với return_exceptions=True, lỗi được trả về trong danh sách thay vì làm dừng tất cả
```

`asyncio.create_task()` chạy một coroutine ở nền mà không có nhóm. Cần giữ tham chiếu tới task, nếu không bộ thu gom rác có thể xóa nó, và lỗi sẽ không ai thấy. Vì vậy với dự án mới, TaskGroup đáng tin cậy hơn.

### Hủy task và timeout

```
async def main():
    task = asyncio.create_task(fetch_rate("USD"))
    task.cancel()                     # bên trong task sẽ phát sinh CancelledError
    try:
        await task
    except asyncio.CancelledError:
        print("đã hủy")

    try:
        async with asyncio.timeout(0.5):   # Python 3.11+
            await fetch_rate("EUR")
    except TimeoutError:
        print("quá lâu")
```

Trong các phiên bản cũ, thay cho `asyncio.timeout()` dùng `asyncio.wait_for(coro, timeout=0.5)`. Còn có `asyncio.wait()`, trả về hai tập hợp: các task đã hoàn thành và chưa hoàn thành; có thể chờ task đầu tiên xong bằng `return_when=asyncio.FIRST_COMPLETED`.

Trong Python 3.15, như đã trình bày trong khóa học, TaskGroup có thêm phương thức `tg.cancel()`: nó hủy toàn bộ task trong nhóm, không cần ngoại lệ phụ trợ. Điều này giải quyết kịch bản «ai trả lời trước»: hỏi ba máy chủ dự phòng, lấy phản hồi đầu tiên, rồi hủy các cái còn lại. Trước đây phải tự ném một ngoại lệ đánh dấu hoặc tự hủy từng task.

## Đồng bộ: Semaphore, Lock, Event, Queue

Asyncio có các nguyên thủy đồng bộ riêng, hoạt động với `await`. Hữu ích nhất là `Semaphore`: giới hạn số thao tác chạy đồng thời, để không làm quá tải API bên ngoài bằng hàng nghìn request.

```
sem = asyncio.Semaphore(10)

async def limited(url):
    async with sem:              # không quá 10 cái cùng lúc
        await asyncio.sleep(0.1) # ở đây sẽ là request thật
        return url
```

- `asyncio.Lock` — chỉ một coroutine được vào đoạn tới hạn.
- `asyncio.Event` — một coroutine chờ qua `await event.wait()`, coroutine khác phát tín hiệu bằng `event.set()`.
- `asyncio.Queue` — hàng đợi kiểu «người sinh — người xử lý»: `await q.put()`, `await q.get()`, `q.task_done()`.

## Lỗi hay gặp: quên await

Lỗi phổ biến nhất là gọi coroutine mà không có `await`:

```
async def main():
    rate = fetch_rate("USD")    # quên await
    print(rate * 2)             # TypeError
# RuntimeWarning: coroutine 'fetch_rate' was never awaited
```

Python không thực thi hàm, mà chỉ tạo đối tượng coroutine. Thấy «was never awaited» trong log thì hãy tìm `await` bị thiếu. Lỗi thứ hai hay gặp là lời gọi chặn (`time.sleep`, `requests.get`) bên trong coroutine. Với HTTP, dùng thư viện bất đồng bộ (httpx, aiohttp), còn đoạn code đồng bộ không tránh được thì chuyển sang luồng bằng `await asyncio.to_thread(func, arg)`.

## Đa luồng trong Python: module threading

Luồng (thread) là một dòng thực thi độc lập trong cùng một tiến trình, các luồng dùng chung bộ nhớ. Với chờ mạng, luồng hoạt động rất tốt và không đòi hỏi viết lại code thành async.

```
import threading, time

def download(name):
    time.sleep(1)               # chờ mạng
    print(name, "xong")

threads = [threading.Thread(target=download, args=(n,)) for n in ("a", "b", "c")]
for t in threads:
    t.start()
for t in threads:
    t.join()                    # tất cả trong khoảng 1 giây
```

Bộ nhớ dùng chung vừa là lợi thế vừa là nguồn lỗi: nếu hai luồng cùng sửa một biến, cần `threading.Lock`. Thường thì không nên tự quản lý luồng bằng tay, mà dùng pool.

### ThreadPoolExecutor

`concurrent.futures.ThreadPoolExecutor` tạo một pool luồng và phân phối công việc cho chúng. `map` trả về kết quả theo đúng thứ tự ban đầu.

```
from concurrent.futures import ThreadPoolExecutor
import time

def fetch(url):
    time.sleep(1)
    return f"{url}: 200"

urls = [f"https://example.com/{i}" for i in range(5)]
with ThreadPoolExecutor(max_workers=5) as pool:
    for result in pool.map(fetch, urls):
        print(result)           # tất cả trong khoảng 1 giây
```

## GIL là gì và vì sao luồng không tăng tốc tính toán

GIL (Global Interpreter Lock) là khóa toàn cục của trình thông dịch CPython. Tại mỗi thời điểm, chỉ một luồng thực thi bytecode Python. Khi luồng chờ nhập/xuất, nó nhả GIL, nên luồng hữu ích cho mạng. Nhưng với tính toán thuần túy, bốn luồng không nhanh hơn một luồng: chúng lần lượt chuyền khóa cho nhau.

```
from concurrent.futures import ThreadPoolExecutor

def heavy(n):
    return sum(i * i for i in range(n))

with ThreadPoolExecutor(4) as pool:
    list(pool.map(heavy, [5_000_000] * 4))   # gần như tuần tự
```

Các thư viện viết bằng C (NumPy và tương tự) thường nhả GIL trong lúc thực hiện các phép toán nặng, nên với chúng luồng vẫn giúp tăng tốc.

## multiprocessing: song song thực sự

Mỗi tiến trình có trình thông dịch riêng và GIL riêng, nên các tiến trình tính toán đồng thời trên nhiều nhân. Cái giá là bộ nhớ riêng: dữ liệu giữa các tiến trình được truyền bằng serialize (pickle), và khởi động một tiến trình tốn kém hơn một luồng.

```
from concurrent.futures import ProcessPoolExecutor

def heavy(n):
    return sum(i * i for i in range(n))

if __name__ == "__main__":      # bắt buộc trên Windows và macOS
    with ProcessPoolExecutor() as pool:
        print(list(pool.map(heavy, [5_000_000] * 4)))
```

Nếu thiếu kiểm tra `if __name__ == "__main__"`, các tiến trình con khi import module sẽ bắt đầu tạo thêm tiến trình mới. Ngoài ra còn có module `multiprocessing` cấp thấp với `Process`, `Pool`, `Queue`, nhưng `ProcessPoolExecutor` đơn giản hơn và có cùng giao diện với pool luồng.

## Chọn asyncio, threading hay multiprocessing

| Bài toán | Công cụ | Lý do |
| --- | --- | --- |

| Tải 200 trang, nhiều request tới API | asyncio (httpx/aiohttp) hoặc ThreadPoolExecutor | chờ mạng, CPU rảnh |

| Nhiều request trong chương trình đã là async | TaskGroup | code đã ở dạng async |

| Thư viện đồng bộ không có bản async | luồng, `asyncio.to_thread` | không cần viết lại code |

| Tính toán nặng bằng Python thuần | ProcessPoolExecutor | vượt GIL, dùng mọi nhân |

## Python không GIL và subinterpreter: có gì mới

Từ Python 3.13 có một bản build không GIL riêng (free-threading, PEP 703) — được đánh dấu bằng chữ t trong tên, ví dụ `python3.15t`. Ở 3.13 nó còn thử nghiệm, từ 3.14 (PEP 779) được hỗ trợ chính thức. Trong bản này các luồng tính toán song song trên nhiều nhân. Nhưng mặc định bạn vẫn nhận Python thông thường có GIL: code đơn luồng trên bản free-threaded chạy hơi chậm hơn, và không phải thư viện nào cũng đã sẵn sàng cho nó.

```
import sys
print(sys._is_gil_enabled())   # False trên bản build free-threaded
```

Con đường thứ hai cho tính toán song song xuất hiện trong 3.14 (PEP 734) — subinterpreter trong module `concurrent.interpreters`: nhiều trình thông dịch cô lập trong cùng một tiến trình, mỗi cái có GIL riêng. Tiện nhất là dùng qua `InterpreterPoolExecutor`, có giao diện giống pool luồng. Trong phép đo của khóa học, bốn tác vụ nặng mất 0.39 s so với 1.03 s khi chạy trên một luồng.

## Tổng kết

Asyncio trong Python là công cụ cho các tác vụ chờ: coroutine khai báo bằng `async def`, tạm dừng bằng `await`, chạy đồng thời qua TaskGroup hoặc `gather`. Luồng giải quyết cùng bài toán mà không cần viết lại code thành async, còn với tính toán nặng cần process, vì GIL ngăn các luồng của CPython thông thường tính song song. Free-threading và subinterpreter đang dần thay đổi bức tranh này, nhưng quy tắc cơ bản «chờ thì dùng asyncio hoặc luồng, tính thì dùng process» vẫn đúng. Trong khóa học, tất cả được thực hành trên ví dụ trình theo dõi chi tiêu: việc lấy tỷ giá tiền tệ chạy song song và mất một giây thay vì ba giây.

## FAQ

**Có thể gọi hàm async từ code thông thường không?**

Có, bằng asyncio.run(coro()): nó tạo event loop, chạy coroutine và đóng loop. Bên trong một event loop đang chạy thì không gọi asyncio.run được — ở đó dùng await.

**Asyncio có chạy trên nhiều luồng không?**

Không, event loop chạy tất cả coroutine trong một luồng và chỉ chuyển đổi tại các điểm await. Vì vậy xung đột dữ liệu ít hơn so với luồng, nhưng một thao tác chặn sẽ làm đứng tất cả.

**Có cần cài asyncio bằng pip không?**

Không, asyncio nằm trong thư viện chuẩn của Python 3; chỉ cần viết import asyncio. Gói asyncio trên PyPI đã cũ và không cần cài.

**Làm sao chạy một hàm nặng từ asyncio mà không chặn event loop?**

Dùng loop.run_in_executor với ProcessPoolExecutor cho tính toán, hoặc asyncio.to_thread cho I/O chặn; event loop vẫn phục vụ các tác vụ khác.

**Coroutine khác task ở điểm nào?**

Coroutine là một đối tượng không làm gì cho đến khi được await. Task bao bọc coroutine và lên lịch nó ngay trên event loop, nên nó chạy song song với phần code khác.


---
Source: Skillaria.Top (https://skillaria.top)