Data API 6 — Async, hiệu năng & dữ liệu lớn

Sau bài auth & bảo mật, service của ta đã đúng chức năng và an toàn. Bài này trả lời câu hỏi vận hành: làm sao endpoint không sập khi phải xuất hàng triệu dòng giao dịch, và làm sao một tiến trình phục vụ được hàng nghìn request đồng thời. Bối cảnh xuyên suốt là hai endpoint khó của NCB: (1) xuất lịch sử giao dịch lớn (khách hàng doanh nghiệp có hàng trăm nghìn giao dịch/năm), và (2) endpoint tổng hợp nặng (báo cáo số dư, thống kê theo chi nhánh). Hiểu sai mô hình async ở đây có thể khiến toàn bộ API đứng hình vì một request duy nhất.

1. Mô hình async: event loop và ASGI

FastAPI đứng trên Starlette, một framework ASGI (Asynchronous Server Gateway Interface). Khác với WSGI cũ (mỗi request một thread, chặn cho tới khi xong), ASGI chạy trên một event loop duy nhất (thư viện asyncio). Event loop là một vòng lặp: nó chạy một coroutine tới khi coroutine đó await một thao tác I/O (đọc DB, gọi HTTP, đọc socket), tại điểm await nó nhả quyền điều khiển để chạy coroutine khác, rồi quay lại khi I/O xong.

Điểm cốt lõi: một thread, nhiều request xen kẽ. Trong lúc request A chờ DB, event loop không ngồi không mà xử lý request B. Nhờ vậy một tiến trình async phục vụ được rất nhiều kết nối đồng thời miễn là phần lớn thời gian là chờ I/O. Đây là lý do async cực hợp với data API: hầu hết công việc là chờ database trả kết quả.

2. async def vs def — bẫy chặn event loop

FastAPI cho cả hai kiểu handler, và cách nó chạy chúng khác hẳn nhau:

Cơ chế threadpool cho def chính là "phao cứu sinh": vì hàm def được đẩy sang một thread khác, nó không chặn event loop.

Bẫy nguy hiểm nhất là đặt code chặn (blocking) bên trong async def. Khi đó nó chạy thẳng trên event loop và không có điểm await nào để nhả — toàn bộ server đứng cho tới khi hàm xong:

# SAI: query đồng bộ trong async def → chặn CẢ server
@app.get("/report")
async def report_bad():
    rows = psycopg2_query("SELECT ...")   # blocking, không await → nghẽn loop
    return rows

# ĐÚNG cách 1: dùng def thường → chạy trong threadpool
@app.get("/report")
def report_ok():
    rows = psycopg2_query("SELECT ...")   # blocking nhưng ở thread riêng
    return rows

# ĐÚNG cách 2: dùng client async thực thụ
@app.get("/report")
async def report_async(pool):
    async with pool.acquire() as conn:
        return await conn.fetch("SELECT ...")   # await → nhả loop khi chờ

Còn một bẫy thứ hai: CPU nặng. Threadpool giải quyết I/O chặn, nhưng với công việc ngốn CPU (parse file lớn, tính toán tổng hợp bằng pandas trên hàng triệu dòng) thì GIL của Python khiến ngay cả threadpool cũng không song song hóa thật. Với CPU-bound, giải pháp là đẩy ra process riêng (ProcessPoolExecutor, hoặc tốt hơn: một hàng đợi tác vụ như Celery/RQ) — xem thêm concurrency & performance trong Python. Ranh giới: async def cho I/O có await; def cho I/O chặn ngắn; process/worker riêng cho CPU nặng và tác vụ dài.

Concurrency vs parallelism (ngắn)

• Concurrency: nhiều việc "cùng tiến triển" bằng cách xen kẽ trên một lõi — đây là những gì event loop làm. Tuyệt vời cho I/O-bound.
• Parallelism: nhiều việc chạy thật sự đồng thời trên nhiều lõi/tiến trình. Cần cho CPU-bound.

Async trong FastAPI cho bạn concurrency, không phải parallelism. Muốn dùng nhiều lõi CPU cho request, bạn tăng số worker (mục 9), mỗi worker là một tiến trình có event loop riêng.

3. Client async cho DB và HTTP

Để async def phát huy tác dụng, mọi I/O bên trong phải là async thật.

Database: asyncpg là driver PostgreSQL async hiệu năng cao. Dùng connection pool để tái sử dụng kết nối (mở kết nối mới rất tốn):

import asyncpg

pool = await asyncpg.create_pool(dsn, min_size=5, max_size=20)

async def get_balance(account_id: int):
    async with pool.acquire() as conn:           # mượn 1 kết nối từ pool
        row = await conn.fetchrow(
            "SELECT balance FROM accounts WHERE id = $1", account_id
        )
        return row["balance"]

min_size/max_size giới hạn số kết nối tới DB — quan trọng vì PostgreSQL có trần max_connections; pool chặn việc mỗi request mở một kết nối gây cạn tài nguyên DB (chi tiết ở database integration).

HTTP client: khi endpoint gọi service ngoài (tỷ giá, dịch vụ định danh), dùng httpx.AsyncClient thay cho requests (đồng bộ). Nên tạo một client dùng chung suốt vòng đời app để tận dụng keep-alive:

import httpx

client = httpx.AsyncClient(timeout=5.0)          # tạo 1 lần, tái dùng

async def fetch_fx(pair: str):
    r = await client.get(f"https://fx.internal/rate/{pair}")
    r.raise_for_status()
    return r.json()

Chủ đề nối DB async trong Python có bài riêng: SQL, API & Async trong Python.

4. BackgroundTasks — làm việc sau khi trả response

Có những việc không cần client chờ: ghi log kiểm toán, gửi email, cập nhật cache. BackgroundTasks cho phép trả response ngay rồi chạy tác vụ sau:

from fastapi import BackgroundTasks

@app.post("/exports")
def create_export(account_id: int, tasks: BackgroundTasks):
    tasks.add_task(write_audit_log, account_id, action="export")
    return {"status": "accepted"}      # trả ngay, log ghi sau

Lưu ý ranh giới: BackgroundTasks chạy trong cùng tiến trình sau khi response gửi đi. Nó hợp cho việc ngắn, phụ trợ. Với việc dài hoặc nặng (sinh file export vài trăm MB, gửi hàng loạt), đừng dùng nó — hãy đẩy sang task queue (Celery/RQ) và trả về một job_id để client hỏi trạng thái sau. Nếu tiến trình restart, BackgroundTasks đang chạy dở sẽ mất; queue có độ bền cao hơn.

5. Caching — đừng tính lại cái không đổi

Endpoint tổng hợp nặng (ví dụ "tổng số dư theo chi nhánh") có thể tốn vài giây mỗi lần. Nếu kết quả chỉ đổi mỗi giờ, cache là đòn bẩy lớn nhất.

Khi nào nên cache kết quả truy vấn nặng: dữ liệu đọc-nhiều/ghi-ít, chấp nhận trễ một chút (dashboard, báo cáo tổng hợp), truy vấn tốn kém. Không cache dữ liệu phải chính xác tức thời (số dư khả dụng để chuyển tiền).

In-memory (dict, functools.lru_cache) đơn giản nhưng chỉ sống trong một tiến trình — nhiều worker sẽ có cache lệch nhau. Redis là cache chia sẻ, dùng chung cho mọi worker và có TTL:

import json

async def branch_summary(branch_id: int):
    key = f"summary:branch:{branch_id}"
    cached = await redis.get(key)
    if cached:
        return json.loads(cached)                 # cache hit
    data = await run_heavy_aggregate(branch_id)   # cache miss → tính
    await redis.set(key, json.dumps(data), ex=3600)  # TTL 1 giờ
    return data

Cache phía HTTP dùng header, để trình duyệt/CDN không gọi lại server:

• Cache-Control: max-age=300 — client được cache 5 phút.
• ETag: server trả một "vân tay" của nội dung (ví dụ hash); lần sau client gửi If-None-Match: <etag>, nếu chưa đổi server trả 304 Not Modified (không body) — tiết kiệm băng thông.

from fastapi import Response, Request

@app.get("/summary/{branch_id}")
async def summary(branch_id: int, request: Request, response: Response):
    data = await branch_summary(branch_id)
    etag = compute_etag(data)                     # minh hoạ: hash nội dung
    if request.headers.get("if-none-match") == etag:
        return Response(status_code=304)
    response.headers["ETag"] = etag
    response.headers["Cache-Control"] = "max-age=300"
    return data

6. Trả DỮ LIỆU LỚN: StreamingResponse & generator

Đây là phần quan trọng nhất cho bài toán xuất giao dịch. Cách ngây thơ — rows = query_all(); return rows — nạp toàn bộ kết quả vào RAM rồi mới serialize. Với 2 triệu giao dịch, tiến trình có thể ngốn vài GB và bị OOM (out of memory) kill.

Giải pháp: StreamingResponse cùng một generator đọc DB theo lô và đẩy từng phần ra socket. RAM chỉ giữ một lô tại một thời điểm.

Xuất NDJSON (mỗi dòng một JSON — dễ xử lý bằng công cụ dữ liệu) theo lô:

import json
from fastapi.responses import StreamingResponse

async def stream_ndjson(account_id: int):
    async with pool.acquire() as conn:
        async with conn.transaction():
            # server-side cursor: DB gửi từng phần, không nạp hết
            async for row in conn.cursor(
                "SELECT id, amount, kind, created_at "
                "FROM transactions WHERE account_id = $1 "
                "ORDER BY id",
                account_id,
            ):
                yield json.dumps(dict(row), default=str) + "\n"

@app.get("/accounts/{account_id}/export")
async def export_txn(account_id: int):
    return StreamingResponse(
        stream_ndjson(account_id),
        media_type="application/x-ndjson",
        headers={"Content-Disposition": "attachment; filename=txn.ndjson"},
    )

Xuất CSV tương tự, chỉ đổi cách format từng dòng và ghi dòng header trước:

import csv, io

async def stream_csv(account_id: int):
    buf = io.StringIO()
    writer = csv.writer(buf)
    writer.writerow(["id", "amount", "kind", "created_at"])   # header
    yield buf.getvalue(); buf.seek(0); buf.truncate(0)
    async with pool.acquire() as conn:
        async for row in conn.cursor("SELECT id, amount, kind, created_at "
                                     "FROM transactions WHERE account_id=$1 "
                                     "ORDER BY id", account_id):
            writer.writerow([row["id"], row["amount"], row["kind"], row["created_at"]])
            yield buf.getvalue(); buf.seek(0); buf.truncate(0)

Ưu điểm phụ: client nhận byte đầu tiên gần như tức thì (Time To First Byte thấp), không phải chờ server gom xong cả file.

Phân trang keyset cho API "lật trang"

Streaming hợp cho tải cả file một lần. Nhưng API phân trang cho UI thì OFFSET lớn lại chậm — DB vẫn phải quét và bỏ qua toàn bộ dòng trước offset. Với dữ liệu lớn, dùng keyset pagination (seek method): lấy trang tiếp theo dựa trên khóa của dòng cuối trang trước, thay vì đếm offset (đã nhắc ở database integration):

-- ▶ Chạy được
SELECT id, amount, kind, created_at
FROM transactions
WHERE account_id = 1 AND id < 5000
ORDER BY id DESC
LIMIT 20;

WHERE id < 5000 dùng chỉ mục để nhảy thẳng tới vị trí, thời gian không đổi dù ở trang 1 hay trang 10.000 — trong khi OFFSET 200000 buộc DB quét 200.000 dòng. Client gửi id nhỏ nhất của trang hiện tại làm con trỏ cho trang sau.

7. Nén gzip

Dữ liệu bảng và JSON nén rất tốt (thường giảm 70–90% kích thước). Bật GZipMiddleware để nén response vượt ngưỡng, tiết kiệm băng thông cho client xa (chi nhánh tỉnh):

from fastapi.middleware.gzip import GZipMiddleware

app.add_middleware(GZipMiddleware, minimum_size=1000)   # chỉ nén body > 1KB

Lưu ý: nén tốn CPU, và middleware nén có thể phải gom nội dung trước khi nén, làm mất lợi thế streaming từng-byte. Với endpoint streaming file rất lớn, cân nhắc để client tự yêu cầu nén hoặc để tầng reverse proxy (nginx) lo, thay vì nén trong app.

8. Giới hạn kích thước & timeout

Không giới hạn là mời gọi sự cố. Cần đặt hàng rào ở nhiều tầng:

• Kích thước request body: chặn upload khổng lồ (thường cấu hình ở reverse proxy, ví dụ nginx client_max_body_size).
• Trần phân trang: limit ≤ 100 (như bài FastAPI căn bản) để không ai kéo cả bảng qua đường phân trang.
• Timeout gọi ra ngoài: httpx.AsyncClient(timeout=5.0) — không để một service ngoài treo giữ kết nối vô hạn.
• Timeout query: đặt statement_timeout cho DB để một truy vấn hỏng không giam kết nối trong pool.
• Timeout tổng của request: cấu hình ở tầng server/proxy để trả 504 thay vì để client treo.

9. Đo & tối ưu

Không đo thì đừng tối ưu. Vài điểm cần nhắm:

• Đo latency: dùng middleware ghi thời gian xử lý, đẩy vào log/metrics (Prometheus). Nhìn p95/p99, không chỉ trung bình — đuôi chậm mới là thứ khách hàng cảm nhận.
• N+1 query: bẫy kinh điển — lấy 100 giao dịch rồi lặp query tên khách hàng cho từng dòng (1 + 100 query). Thay bằng một query có JOIN. Đây là nguyên nhân chậm phổ biến nhất trong data API.

-- ▶ Chạy được
SELECT t.id, t.amount, c.full_name
FROM transactions t
JOIN accounts a ON a.id = t.account_id
JOIN customers c ON c.id = a.customer_id
WHERE a.id = 1
ORDER BY t.created_at DESC
LIMIT 20;

• Connection pool: pool quá nhỏ → request xếp hàng chờ kết nối; quá lớn → vượt max_connections của PostgreSQL. Đo tỷ lệ chờ pool để chỉnh max_size.
• Số uvicorn worker: mỗi worker là một tiến trình có event loop riêng, dùng được thêm một lõi CPU. Công thức khởi điểm phổ biến cho workload trộn I/O + chút CPU là (2 × số_lõi) + 1; workload thuần I/O có thể ít worker hơn (async đã cho concurrency), workload CPU nặng thì bám sát số lõi. Luôn đo lại dưới tải thật.

# minh hoạ: 4 lõi → thử 9 worker rồi đo lại
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 9

Trong production thường chạy uvicorn dưới một process manager (gunicorn với uvicorn worker, hoặc nhiều pod trong k8s), chi tiết ở bài triển khai và tinh thần chất lượng vận hành ở production quality.

Use case thực tế

Bài toán: Khách hàng doanh nghiệp yêu cầu tải toàn bộ lịch sử giao dịch năm 2025 (≈ 1,8 triệu dòng) từ Internet Banking để đối soát. Endpoint cũ return list(query_all()) nạp hết vào RAM: tiến trình phình lên ~2,4 GB, chờ ~40 giây rồi thỉnh thoảng bị OOM kill, kéo theo các request khác trên cùng worker.

Xử lý:

1. Đổi sang StreamingResponse + server-side cursor đọc DB theo lô 5.000 dòng, xuất NDJSON. RAM của tiến trình chỉ còn dao động ~60–80 MB vì mỗi thời điểm chỉ giữ một lô.
2. Query dùng chỉ mục trên (account_id, id) và ORDER BY id để cursor chạy tuần tự, không phát sinh sort tốn bộ nhớ.
3. Time To First Byte giảm từ ~40 giây xuống dưới 1 giây — client thấy file bắt đầu tải ngay.
4. Với màn hình lật trang (không phải tải cả file), chuyển từ OFFSET sang keyset: trang 9.000 trước đây mất ~1,2 giây (DB quét 180.000 dòng) nay còn ~10 ms nhờ WHERE id < :cursor bám chỉ mục.
5. Endpoint báo cáo tổng hợp theo chi nhánh (chạy ~3 giây) được cache Redis TTL 1 giờ; sau lần đầu, các lần sau trả trong ~5 ms và không còn đè tải lên DB vào giờ cao điểm.
6. Bật statement_timeout và timeout HTTP để một truy vấn hỏng không giam kết nối pool.

Kết quả: một worker giờ phục vụ đồng thời nhiều lượt export mà không OOM; DB nhẹ tải hơn nhờ cache và keyset; trải nghiệm tải file mượt vì dữ liệu chảy dần thay vì chờ một cục lớn.

Ghi nhớ

• FastAPI/Starlette là ASGI chạy trên event loop một thread; nhiều request xen kẽ tại các điểm await I/O → concurrency cao cho workload I/O-bound.
• async def chạy trên loop (phải await client async); def thường chạy trong threadpool (an toàn cho I/O chặn). Không đặt code chặn hay CPU nặng trong async def — sẽ nghẽn cả server.
• CPU nặng cần process/worker riêng (queue), không phải threadpool; async = concurrency, tăng worker = parallelism.
• Dùng client async thật (asyncpg + connection pool, httpx.AsyncClient tái dùng) để async def có ý nghĩa.
• BackgroundTasks cho việc phụ ngắn sau response; việc dài/nặng đẩy sang task queue.
• Cache (Redis chia sẻ, in-memory cục bộ; Cache-Control/ETag+304) cho truy vấn nặng đọc-nhiều-ghi-ít; không cache dữ liệu cần tức thời.
• Dữ liệu lớn: StreamingResponse + generator đọc DB theo lô (CSV/NDJSON), tránh nạp hết vào RAM; UI lật trang dùng keyset thay OFFSET lớn.
• Đặt hàng rào: nén gzip (cẩn thận khi streaming), giới hạn body/limit, timeout HTTP/DB; đo p95/p99, diệt N+1 bằng JOIN, chỉnh pool và số worker theo đo đạc thực tế.