Xây dựng Background Worker hiệu năng cao trong Python với Celery và Redis

Table of Contents

Bắt đầu nhanh (5 phút)

Việc làm tắc nghẽn (blocking) vòng lặp xử lý web request sẽ phá hỏng trải nghiệm người dùng. Nếu một endpoint phải tạo báo cáo PDF dài 40 trang hoặc gọi một cổng thanh toán chậm chạp, thời gian phản hồi có thể dễ dàng tăng vọt từ 45ms lên hơn 4 giây. Đẩy các tác vụ nặng sang các background worker sẽ giúp API của bạn luôn phản hồi nhanh chóng.

Celery và Redis đảm nhận việc phân chia tác vụ này rất gọn gàng. Khởi chạy một Redis broker trên máy cục bộ bằng Docker để bắt đầu:

docker run -d --name redis-broker -p 6379:6379 redis:7-alpine

Cài đặt các thư viện Python cần thiết:

pip install celery redis

Định nghĩa background task trong file tasks.py:

from celery import Celery
import time

app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1"
)

@app.task
def process_signup_email(user_email: str) -> str:
    time.sleep(2)  # Giả lập độ trễ của SMTP server
    return f"Đã gửi email chào mừng tới {user_email}"

Khởi chạy tiến trình worker trong terminal của bạn:

celery -A tasks worker --loglevel=info --concurrency=2

Đưa job đầu tiên vào hàng đợi từ Python REPL:

>>> from tasks import process_signup_email
>>> async_result = process_signup_email.delay("[email protected]")
>>> async_result.id
'2e9564f2-9860-498c-85a7-96a24eb5ef74'
>>> async_result.ready()
False
>>> async_result.get(timeout=5)
'Đã gửi email chào mừng tới [email protected]'

Phía gọi (caller) bàn giao payload chỉ trong vài mili-giây, nhận lại một UUID và giải phóng ngay lập tức HTTP thread.

Phân tích chuyên sâu

Để mở rộng quy mô Celery, bạn cần hiểu cách client, message transport và các tiến trình hệ điều hành phối hợp xử lý khối lượng công việc.

Message broker so với result backend

Celery tách biệt hoàn toàn logic xếp hàng đợi khỏi việc theo dõi kết quả đầu ra:

Broker (Redis DB 0): Đóng vai trò là hàng đợi truyền tải (transport queue). Khi gọi delay(), các đối số của task sẽ được serialize thành chuỗi JSON và đẩy vào một Redis list. Web worker ngắt kết nối ngay lập tức.
Result Backend (Redis DB 1): Theo dõi trạng thái thực thi (PENDING, SUCCESS, FAILURE) và giá trị trả về. Nếu bạn chỉ quan tâm đến các tác vụ phụ (side effects)—chẳng hạn như gửi email giao dịch—hãy thiết lập ignore_result=True. Việc tắt backend sẽ xóa bỏ hoàn toàn các Redis key tạm thời và giảm tới 50% IOPS ghi.

Các mô hình concurrency: prefork so với gevent

Pool mặc định prefork dựa trên module multiprocessing của Python. Vì mỗi tiến trình worker có không gian bộ nhớ riêng, prefork giúp vượt qua Global Interpreter Lock (GIL) đối với các tác vụ CPU-bound như nén ảnh hay phân tích dữ liệu. Tuy nhiên, các tiến trình OS khá nặng. Việc khởi tạo 32 prefork worker có thể tiêu tốn từ 1.5 GB đến 3 GB RAM.

Nếu khối lượng công việc chủ yếu là network I/O—như gọi định kỳ các REST API bên ngoài hoặc cào dữ liệu web (scraping)—hãy chuyển sang event loop. Một tiến trình worker gevent duy nhất chỉ dùng chưa đến 80 MB RAM nhưng có thể quản lý hàng trăm kết nối đồng thời:

celery -A tasks worker --pool=gevent --concurrency=100 --loglevel=info

Task acknowledgment và tính idempotency

Theo mặc định, Celery xác nhận tin nhắn sớm. Broker sẽ đánh dấu task hoàn thành ngay thời điểm worker lấy task ra khỏi hàng đợi (dequeue). Nếu Linux buộc dừng (kill) tiến trình worker do tràn bộ nhớ (OOM) giữa lúc đang xử lý, task đó sẽ biến mất không dấu vết.

Hãy thiết lập task_acks_late = True để giữ lại acknowledgment cho đến khi hàm kết thúc thành công. Tuy nhiên, điều này đi kèm một sự đánh đổi. Nếu một node bị crash giữa chừng, Redis sẽ gửi lại job cho một worker khác. Hãy thiết kế mọi thao tác thay đổi cơ sở dữ liệu và lệnh gọi thanh toán có tính idempotent bằng cách sử dụng các idempotency key duy nhất trước khi bật tùy chọn này.

Kỹ thuật nâng cao

Điều phối pipeline với Canvas

Các hệ thống production thường đòi hỏi những quy trình gồm nhiều bước. Bạn có thể cần trừ tiền thẻ, lưu hóa đơn vào PostgreSQL và gửi push notification theo thứ tự tuần tự. Các chain và group trong Celery Canvas kết hợp các pipeline này một cách gọn gàng mà không cần lồng các callback phức tạp:

from celery import Celery, chain, group

app = Celery("ecommerce", broker="redis://localhost:6379/0", backend="redis://localhost:6379/1")

@app.task
def charge_customer(order_id: int, amount: float):
    return {"order_id": order_id, "transaction_id": f"tx_{order_id}", "amount": amount}

@app.task
def generate_invoice(charge_data: dict):
    invoice_number = f"INV-{charge_data['order_id']}"
    return {**charge_data, "invoice": invoice_number}

@app.task
def notify_customer(invoice_data: dict):
    return f"Đã gửi thông báo cho hóa đơn {invoice_data['invoice']}"

@app.task
def sync_inventory_item(item_id: int):
    return f"Đã giảm tồn kho cho sản phẩm {item_id}"

# Pipeline tuần tự: toán tử pipe truyền output sang task tiếp theo
workflow = chain(
    charge_customer.s(order_id=4021, amount=99.50) |
    generate_invoice.s() |
    notify_customer.s()
)
workflow.apply_async()

# Thực thi song song: chạy các job độc lập đồng thời
inventory_sync = group(
    sync_inventory_item.s(item_id=101),
    sync_inventory_item.s(item_id=102),
    sync_inventory_item.s(item_id=103)
)
inventory_sync.apply_async()

Giới hạn tốc độ (Rate limiting) cho các tích hợp bên ngoài

Các API bên thứ ba thường áp dụng hạn ngạch rate limit nghiêm ngặt. Ví dụ: Twilio sẽ từ chối các yêu cầu gửi SMS vượt quá 10 tin nhắn/giây đối với các đầu số tiêu chuẩn. Celery cung cấp sẵn bộ giới hạn tốc độ theo thuật toán token-bucket ngay ở cấp định nghĩa task:

@app.task(rate_limit="10/s", max_retries=3, default_retry_delay=5)
def send_sms_alert(phone_number: str, message: str):
    try:
        # Gửi SMS qua nhà cung cấp dịch vụ bên ngoài
        pass
    except ExternalServiceError as exc:
        # Thử lại với exponential backoff: 5s, 10s, 20s
        countdown = 5 * (2 ** send_sms_alert.request.retries)
        raise send_sms_alert.retry(exc=exc, countdown=countdown)

Hãy lưu ý rằng Celery áp dụng rate_limit trên từng worker instance thay vì trên toàn bộ cluster của bạn. Nếu bạn chạy 10 container worker riêng biệt với rate_limit="10/s", cluster của bạn có thể gửi tới nhà cung cấp tới 100 request mỗi giây. Hãy sử dụng Redis token bucket với thao tác atomic hoặc dùng một proxy nếu cần đảm bảo giới hạn toàn cục tuyệt đối.

Kinh nghiệm thực tế

1. Cô lập các hàng đợi theo mức độ ưu tiên

Việc đặt mã xác thực tức thời và tác vụ xuất báo cáo nặng kéo dài 2 giờ chung vào một hàng đợi mặc định chắc chắn sẽ dẫn đến nghẽn hệ thống. Một đợt bùng nổ 5.000 tác vụ xuất PDF sẽ chiếm trọn mọi worker slot. Khi đó, email đặt lại mật khẩu sẽ phải nằm chờ dài trong hàng đợi tồn đọng (backlog).

Hãy phân chia các task vào các hàng đợi riêng biệt trong cấu hình định tuyến (routing):

app.conf.task_routes = {
    "tasks.send_sms_alert": {"queue": "priority"},
    "tasks.generate_invoice": {"queue": "billing"},
    "tasks.export_large_csv": {"queue": "batch"},
}

Chạy các tiến trình worker riêng biệt dành riêng cho từng hàng đợi:

# Worker ưu tiên xử lý thông lượng cao
celery -A tasks worker -Q priority -c 4

# Worker chuyên dụng cho tác vụ theo đợt (tránh để job nặng chiếm hết tài nguyên của hàng đợi API)
celery -A tasks worker -Q batch -c 1

2. Thiết lập giới hạn bộ nhớ cho worker

Các thư viện như Pandas, ReportLab hay Pillow thường xuyên gây phân mảnh bộ nhớ (memory fragmentation) trong các tiến trình chạy lâu dài. Các tiến trình con sẽ phình to dần cho đến khi Linux kích hoạt tín hiệu SIGKILL. Hãy yêu cầu Celery khởi động lại các worker con sau khi xử lý một lượng công việc nhất định:

# Khởi động lại tiến trình con sau 500 task hoặc khi sử dụng 200MB bộ nhớ
app.conf.worker_max_tasks_per_child = 500
app.conf.worker_max_memory_per_child = 200000  # 200,000 KB (~195 MB)

3. Giảm prefetch multiplier cho các job không đồng đều

Mặc định, Celery lấy các task theo từng đợt (batch). Với prefetch_multiplier bằng 4, một worker có concurrency bằng 4 sẽ lấy trước 16 task. Nếu 4 task trong số đó mỗi task cần 5 phút để xử lý, 12 task còn lại sẽ phải nằm chờ trong bộ đệm cục bộ của worker đó trong khi các worker khác lại đang rảnh rỗi.

Đối với các job chạy lâu hoặc có thời lượng không đồng đều, hãy đặt prefetch về 1 để các task chưa được nhận vẫn nằm trên Redis và sẵn sàng cho các worker khác:

app.conf.worker_prefetch_multiplier = 1
app.conf.task_acks_late = True

4. Thiết lập thời hạn thực thi rõ ràng

Sự cố nghẽn socket không được xử lý trong một HTTP request ngược dòng có thể làm treo worker vô thời hạn. Hãy bảo vệ từng task bằng các mốc giới hạn thời gian (timeout) mềm và cứng rõ ràng:

# Soft timeout kích hoạt exception SoftTimeLimitExceeded có thể bắt được để dọn dẹp tài nguyên
# Hard timeout gửi SIGKILL đến tiến trình con sau đó 10 giây
app.conf.task_soft_time_limit = 120
app.conf.task_time_limit = 130

Cấu hình các giới hạn này sẽ bảo vệ bộ nhớ máy chủ, ngăn chặn tình trạng kẹt worker slot và đảm bảo độ trễ hàng đợi luôn ổn định, dự đoán được trên môi trường production.