Lập Trình Bất Đồng Bộ trong FastAPI: Tăng Hiệu Suất với async/await

Giới Thiệu

Một trong những tính năng mạnh mẽ nhất của FastAPI là hỗ trợ lập trình bất đồng bộ (asynchronous programming) với cú pháp async/await của Python. Khả năng này cho phép bạn xây dựng API hiệu suất cao, có thể xử lý hàng ngàn request đồng thời một cách hiệu quả.

Trong hướng dẫn toàn diện này, chúng ta sẽ khám phá:

Lập trình bất đồng bộ là gì và tại sao nó quan trọng
Cách async/await hoạt động trong FastAPI
Khi nào nên dùng async vs sync endpoints
Best practices và các lỗi thường gặp
Ví dụ thực tế với database và external APIs

Lập Trình Bất Đồng Bộ Là Gì?

Lập trình bất đồng bộ cho phép ứng dụng của bạn xử lý nhiều thao tác đồng thời mà không chặn (block) luồng thực thi. Thay vì đợi các thao tác I/O (như truy vấn database, HTTP request, hoặc đọc file) hoàn thành, ứng dụng có thể chuyển sang xử lý các task khác.

Đồng Bộ vs Bất Đồng Bộ

Đồng bộ (blocking):

def fetch_user(user_id: int):
    # Đợi 2 giây cho database query
    user = database.query(user_id)  # ⏸️ Thread bị block ở đây
    return user
 
def fetch_orders(user_id: int):
    # Đợi 1 giây cho database query
    orders = database.get_orders(user_id)  # ⏸️ Thread bị block ở đây
    return orders
 
# Tổng thời gian: 2s + 1s = 3 giây
user = fetch_user(1)
orders = fetch_orders(1)

Bất đồng bộ (non-blocking):

async def fetch_user(user_id: int):
    # Bắt đầu database query, không đợi
    user = await database.query(user_id)  # 🔄 Có thể xử lý request khác
    return user
 
async def fetch_orders(user_id: int):
    # Bắt đầu database query, không đợi
    orders = await database.get_orders(user_id)  # 🔄 Có thể xử lý request khác
    return orders
 
# Tổng thời gian: max(2s, 1s) = 2 giây (chạy đồng thời!)
user, orders = await asyncio.gather(
    fetch_user(1),
    fetch_orders(1)
)

Hỗ Trợ Async Trong FastAPI

FastAPI hỗ trợ cả synchronous và asynchronous endpoints, cho bạn sự linh hoạt dựa trên use case.

Async Endpoint Cơ Bản

from fastapi import FastAPI
import asyncio
 
app = FastAPI()
 
@app.get("/")
async def root():
    """Async endpoint - có thể xử lý concurrent requests hiệu quả"""
    await asyncio.sleep(1)  # Mô phỏng I/O operation
    return {"message": "Hello World"}

Kết Hợp Sync và Async Endpoints

@app.get("/sync-endpoint")
def sync_endpoint():
    """Synchronous endpoint - chạy trong thread pool"""
    import time
    time.sleep(1)  # Blocking operation
    return {"message": "Sync operation"}
 
@app.get("/async-endpoint")
async def async_endpoint():
    """Asynchronous endpoint - chạy trong event loop"""
    await asyncio.sleep(1)  # Non-blocking operation
    return {"message": "Async operation"}

Cách FastAPI xử lý:

Async endpoints (async def): Chạy trực tiếp trong event loop
Sync endpoints (def): Chạy trong thread pool riêng để tránh blocking

Khi Nào Nên Dùng Async vs Sync

Dùng `async def` khi:

✅ I/O-bound operations:

Truy vấn database (với async drivers)
HTTP requests tới external APIs
File operations (với async libraries)
Cache operations (Redis, Memcached)
Message queue operations

✅ Yêu cầu concurrency cao:

Microservices xử lý nhiều concurrent requests
WebSocket connections
Ứng dụng real-time

Dùng `def` thông thường khi:

✅ CPU-bound operations:

Tính toán nặng
Xử lý dữ liệu
Xử lý ảnh/video
Machine learning inference

✅ Thư viện chỉ hỗ trợ sync:

Database drivers cũ (SQLAlchemy sync mode)
Thư viện third-party không có async support

Quan trọng: Đừng dùng await với sync libraries trong async functions—điều này làm mất tác dụng!

Ví Dụ Thực Tế

Ví Dụ 1: Async Database Operations

Sử dụng SQLAlchemy với async support:

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import select
 
# Thiết lập async database
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)
 
app = FastAPI()
 
# Dependency cho database session
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session
 
# Async endpoint với database query
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    """Fetch user bất đồng bộ"""
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
 
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
 
    return user

Ví Dụ 2: Concurrent External API Calls

import httpx
from fastapi import FastAPI
 
app = FastAPI()
 
@app.get("/weather/{city}")
async def get_weather_and_forecast(city: str):
    """Fetch thời tiết hiện tại và dự báo đồng thời"""
    async with httpx.AsyncClient() as client:
        # Chạy cả 2 API calls đồng thời
        current_weather, forecast = await asyncio.gather(
            client.get(f"https://api.weather.com/current/{city}"),
            client.get(f"https://api.weather.com/forecast/{city}")
        )
 
    return {
        "current": current_weather.json(),
        "forecast": forecast.json()
    }

Lợi ích hiệu suất: Nếu mỗi API mất 500ms, thực thi đồng thời mất ~500ms so với 1000ms nếu chạy tuần tự.

Ví Dụ 3: Background Tasks với Async

from fastapi import BackgroundTasks
import asyncio
 
async def send_email_async(email: str, message: str):
    """Mô phỏng gửi email bất đồng bộ"""
    await asyncio.sleep(2)  # Mô phỏng việc gửi email
    print(f"Email sent to {email}: {message}")
 
@app.post("/register")
async def register_user(
    email: str,
    password: str,
    background_tasks: BackgroundTasks
):
    """Đăng ký user và gửi welcome email trong background"""
    # Lưu user vào database (async)
    user = await create_user_in_db(email, password)
 
    # Thêm email task vào background (không block response)
    background_tasks.add_task(send_email_async, email, "Welcome!")
 
    return {"message": "User registered", "user_id": user.id}

Ví Dụ 4: WebSocket với Async

from fastapi import WebSocket
 
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    """WebSocket connection với async message handling"""
    await websocket.accept()
 
    try:
        while True:
            # Đợi message từ client (non-blocking)
            data = await websocket.receive_text()
 
            # Xử lý message (có thể là async database query)
            response = await process_message(data)
 
            # Gửi response về client
            await websocket.send_text(f"Response: {response}")
    except:
        await websocket.close()
 
async def process_message(message: str):
    """Xử lý message bất đồng bộ"""
    await asyncio.sleep(0.1)  # Mô phỏng processing
    return message.upper()

Các Lỗi Thường Gặp và Giải Pháp

Lỗi 1: Dùng Blocking Code trong Async Functions

❌ Sai:

@app.get("/users")
async def get_users():
    # time.sleep() block toàn bộ event loop!
    import time
    time.sleep(5)  # SAI: Block tất cả requests khác
    return {"users": []}

✅ Đúng:

@app.get("/users")
async def get_users():
    # asyncio.sleep() là non-blocking
    await asyncio.sleep(5)  # ĐÚNG: Requests khác vẫn được xử lý
    return {"users": []}

Lỗi 2: Quên `await`

❌ Sai:

@app.get("/user/{user_id}")
async def get_user(user_id: int):
    # Trả về coroutine, không phải kết quả thực tế!
    user = fetch_user_async(user_id)  # Thiếu await
    return user  # Trả về coroutine object, không phải user data

✅ Đúng:

@app.get("/user/{user_id}")
async def get_user(user_id: int):
    user = await fetch_user_async(user_id)  # Đúng cách await kết quả
    return user

Lỗi 3: Trộn Sync và Async Database Drivers

❌ Sai:

from sqlalchemy import create_engine  # Sync engine
 
@app.get("/users")
async def get_users():
    # Dùng sync SQLAlchemy trong async function block event loop
    users = session.query(User).all()  # Blocking call!
    return users

✅ Đúng:

from sqlalchemy.ext.asyncio import create_async_engine
 
@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
    # Dùng async SQLAlchemy
    result = await db.execute(select(User))
    users = result.scalars().all()
    return users

Lỗi 4: Không Dùng Async HTTP Clients

❌ Sai:

import requests  # Sync library
 
@app.get("/external-data")
async def fetch_external():
    # requests.get() là blocking!
    response = requests.get("https://api.example.com/data")
    return response.json()

✅ Đúng:

import httpx  # Async HTTP client
 
@app.get("/external-data")
async def fetch_external():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        return response.json()

Best Practices Về Hiệu Suất

1. Dùng Connection Pooling

from sqlalchemy.ext.asyncio import create_async_engine
 
# Cấu hình connection pool cho hiệu suất tối ưu
engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,        # Số connections tối đa trong pool
    max_overflow=10,     # Connections bổ sung nếu pool đầy
    pool_pre_ping=True,  # Verify connections trước khi dùng
)

2. Đặt Timeout cho External Calls

import httpx
 
async def fetch_with_timeout(url: str):
    """Luôn đặt timeout để tránh requests bị treo"""
    async with httpx.AsyncClient(timeout=5.0) as client:
        try:
            response = await client.get(url)
            return response.json()
        except httpx.TimeoutException:
            return {"error": "Request timed out"}

3. Dùng `asyncio.gather()` cho Concurrent Operations

@app.get("/dashboard/{user_id}")
async def get_dashboard(user_id: int, db: AsyncSession = Depends(get_db)):
    """Fetch nhiều resources đồng thời"""
    user_task = db.execute(select(User).where(User.id == user_id))
    orders_task = db.execute(select(Order).where(Order.user_id == user_id))
    stats_task = fetch_user_stats(user_id)
 
    # Đợi tất cả tasks đồng thời
    user_result, orders_result, stats = await asyncio.gather(
        user_task,
        orders_task,
        stats_task
    )
 
    return {
        "user": user_result.scalar_one(),
        "orders": orders_result.scalars().all(),
        "stats": stats
    }

4. Dùng Semaphores để Giới Hạn Concurrency

# Global semaphore để giới hạn concurrent external API calls
api_semaphore = asyncio.Semaphore(10)  # Tối đa 10 concurrent calls
 
async def fetch_external_data(url: str):
    """Giới hạn concurrent external API calls"""
    async with api_semaphore:
        async with httpx.AsyncClient() as client:
            response = await client.get(url)
            return response.json()

Testing Async Endpoints

Dùng pytest với pytest-asyncio:

import pytest
from httpx import AsyncClient
from main import app
 
@pytest.mark.asyncio
async def test_async_endpoint():
    """Test async endpoint"""
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/users/1")
        assert response.status_code == 200
        assert "user" in response.json()

Monitoring Async Performance

Theo dõi các chỉ số quan trọng:

import time
from fastapi import Request
 
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    """Đo thời gian xử lý request"""
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

Kết Luận

Lập trình bất đồng bộ trong FastAPI mở ra cải thiện hiệu suất đáng kể cho các ứng dụng I/O-bound. Các điểm chính:

Dùng async def cho I/O operations (database, HTTP, files)
Dùng def thông thường cho CPU-bound operations hoặc thư viện chỉ hỗ trợ sync
Luôn await async functions và dùng các thư viện tương thích async
Chạy concurrent operations với asyncio.gather() để tối đa hiệu suất
Đặt timeout và dùng connection pooling cho production
Test kỹ lưỡng với pytest-asyncio

Bằng cách tuân theo các pattern này và tránh các lỗi thường gặp, bạn sẽ xây dựng được các ứng dụng FastAPI scale hiệu quả và xử lý hàng ngàn concurrent users một cách dễ dàng.

Lập Trình Bất Đồng Bộ trong FastAPI: Tăng Hiệu Suất với async/await

Giới Thiệu

Lập Trình Bất Đồng Bộ Là Gì?

Đồng Bộ vs Bất Đồng Bộ

Hỗ Trợ Async Trong FastAPI

Async Endpoint Cơ Bản

Kết Hợp Sync và Async Endpoints

Khi Nào Nên Dùng Async vs Sync

Dùng `async def` khi:

Dùng `def` thông thường khi:

Ví Dụ Thực Tế

Ví Dụ 1: Async Database Operations

Ví Dụ 2: Concurrent External API Calls

Ví Dụ 3: Background Tasks với Async

Ví Dụ 4: WebSocket với Async

Các Lỗi Thường Gặp và Giải Pháp

Lỗi 1: Dùng Blocking Code trong Async Functions

Lỗi 2: Quên `await`

Lỗi 3: Trộn Sync và Async Database Drivers

Lỗi 4: Không Dùng Async HTTP Clients

Best Practices Về Hiệu Suất

1. Dùng Connection Pooling

2. Đặt Timeout cho External Calls

3. Dùng `asyncio.gather()` cho Concurrent Operations

4. Dùng Semaphores để Giới Hạn Concurrency

Testing Async Endpoints

Monitoring Async Performance

Kết Luận

Tài Liệu Tham Khảo

📬 Đăng ký nhận bản tin

💬 Comments

Giới Thiệu

Lập Trình Bất Đồng Bộ Là Gì?

Đồng Bộ vs Bất Đồng Bộ

Hỗ Trợ Async Trong FastAPI

Async Endpoint Cơ Bản

Kết Hợp Sync và Async Endpoints

Khi Nào Nên Dùng Async vs Sync

Dùng async def khi:

Dùng def thông thường khi:

Ví Dụ Thực Tế

Ví Dụ 1: Async Database Operations

Ví Dụ 2: Concurrent External API Calls

Ví Dụ 3: Background Tasks với Async

Ví Dụ 4: WebSocket với Async

Các Lỗi Thường Gặp và Giải Pháp

Lỗi 1: Dùng Blocking Code trong Async Functions

Lỗi 2: Quên await

Lỗi 3: Trộn Sync và Async Database Drivers

Lỗi 4: Không Dùng Async HTTP Clients

Best Practices Về Hiệu Suất

1. Dùng Connection Pooling

2. Đặt Timeout cho External Calls

3. Dùng asyncio.gather() cho Concurrent Operations

4. Dùng Semaphores để Giới Hạn Concurrency

Testing Async Endpoints

Monitoring Async Performance

Kết Luận

Tài Liệu Tham Khảo

📬 Đăng ký nhận bản tin

💬 Comments

Dùng `async def` khi:

Dùng `def` thông thường khi:

Lỗi 2: Quên `await`

3. Dùng `asyncio.gather()` cho Concurrent Operations