Mở đầu

Tập này đưa Celery vào ứng dụng quản lý chi tiêu: việc nặng như gửi báo cáo tháng chạy nền bằng worker, lặp lại đúng lịch bằng beat, quan sát được bằng Flower. Kết thúc tập, bạn có hàng đợi chạy ổn định bằng systemd, task lỗi tự thử lại có giới hạn rồi vào dead-letter. Nối tiếp tập 9, chúng ta tái dùng Redis làm cache để làm broker.

1. Cài đặt và cấu hình Celery

Cài Celery kèm hỗ trợ Redis, rồi tạo celery.py cạnh settings.py để app tự khám phá task.

# config/celery.py
import os
from celery import Celery
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.prod")
app = Celery("expense")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()
# config/settings/prod.py
CELERY_BROKER_URL = env("REDIS_URL")
CELERY_RESULT_BACKEND = env("REDIS_URL")
CELERY_TASK_TIME_LIMIT = 300
CELERY_TASK_SOFT_TIME_LIMIT = 240
CELERY_TASK_ACKS_LATE = True
CELERY_WORKER_PREFETCH_MULTIPLIER = 1

Đáng chú ý: acks_late khiến task chỉ được xác nhận sau khi chạy xong, nên worker chết giữa chừng thì task được giao lại; prefetch_multiplier = 1 ngăn worker ôm nhiều task cùng lúc.

2. Viết task đầu tiên

Task báo cáo tháng gom dữ liệu rồi gửi email, chạy ngoài request, người dùng không phải chờ.

# expenses/tasks.py
from celery import shared_task
from django.contrib.auth import get_user_model
from django.core.mail import send_mail
from .services import monthly_summary
@shared_task(
    autoretry_for=(Exception,),
    retry_backoff=True,
    max_retries=3,
)
def send_monthly_report(user_id):
    user = get_user_model().objects.get(pk=user_id)
    body = monthly_summary(user)
    send_mail("Bao cao chi tieu thang", body,
              "no-reply@example.com", [user.email])

Chỉ truyền khóa chính vào task rồi truy vấn lại bên trong. Task phải idempotent: chạy lại hai lần cũng cho cùng kết quả, vì Celery có thể giao lại task.

3. Lên lịch với Celery Beat

Beat là đồng hồ của hệ thống: bắn task đúng lịch, worker thực thi. Khai báo lịch trong settings bằng crontab.

# config/settings/prod.py
from celery.schedules import crontab
CELERY_BEAT_SCHEDULE = {
    "send-monthly-reports": {
        "task": "expenses.tasks.send_monthly_reports",
        "schedule": crontab(day_of_month=1, hour=7, minute=0),
    },
    "remind-due-bills": {
        "task": "expenses.tasks.remind_due_bills",
        "schedule": crontab(hour=8, minute=0),
    },
}

Task send_monthly_reports duyệt người dùng rồi gọi send_monthly_report.delay(user.id) cho từng người. Chia nhỏ giúp một email lỗi không chặn cả loạt, từng báo cáo retry độc lập.

4. Chạy worker và beat bằng systemd

Trên production, worker và beat là service hệ điều hành, tự khởi động lại khi chết. Mỗi tiến trình một unit systemd riêng, beat chỉ chạy một bản.

# /etc/systemd/system/celery-worker.service
[Unit]
Description=Celery worker - quan ly chi tieu
After=network.target
[Service]
User=deploy
WorkingDirectory=/srv/expense
EnvironmentFile=/srv/expense/.env
ExecStart=/srv/expense/venv/bin/celery -A config worker -l info --concurrency=2
Restart=always
[Install]
WantedBy=multi-user.target
sudo systemctl enable --now celery-worker celery-beat
sudo systemctl status celery-worker

Hai bản beat sẽ bắn trùng task, nên nhiều máy chủ thì chỉ một máy chạy beat, worker nhân bản thoải mái. Log worker xem bằng journalctl -u celery-worker, như cách tập trung log ở tập 7.

5. Giám sát bằng Flower

Flower là giao diện web xem worker, task đang chạy, task thất bại và tốc độ xử lý.

/srv/expense/venv/bin/celery -A config flower --port=5555

Đừng phơi Flower ra Internet: đặt sau Nginx với xác thực cơ bản, như cách khóa endpoint nội bộ ở tập 3. Mỗi sáng nhìn Flower là biết đêm qua bao nhiêu báo cáo thành công, task nào chậm bất thường.

6. Retry có giới hạn và dead-letter

Retry vô hạn làm hàng đợi tắc nghẽn vì task lỗi cứ quay lại, chiếm hết worker. Quy tắc ở đây: thử lại tối đa ba lần, thời gian chờ tăng dần, hết ba lần thì đưa vào dead-letter để xử lý tay.

# expenses/tasks.py
from celery import Task
class DeadLetterTask(Task):
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        self.app.send_task(
            "expenses.tasks.handle_dead_letter",
            args=[self.name, args, kwargs, str(exc)],
            queue="dead_letter",
        )
@shared_task(base=DeadLetterTask, bind=True, max_retries=3)
def remind_due_bills(self):
    try:
        send_due_reminders()
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60 * 2 ** self.request.retries)

Task handle_dead_letter chỉ ghi log và gửi cảnh báo qua Sentry ở tập 7. Hàng đợi dead-letter là nơi duy nhất cần xem để biết việc nào đang kẹt, thay vì mò trong log.

Áp dụng vào dự án của bạn

Bắt đầu từ task thật ít rủi ro: gửi email báo cáo tháng bằng beat vào mùng 1. Khi đã tin tưởng, chuyển dần việc nặng khác sang hàng đợi: nhắc hạn thanh toán, tổng hợp chi tiêu, xuất CSV. Quy tắc giữ lại: request chỉ làm việc nhanh, thứ chậm hơn vài trăm mili giây đều xứng đáng thành task.

Bài viết liên quan

Câu hỏi thường gặp

Dùng chung Redis cho cache và broker có sao không?

Được, nhưng tách database index khác nhau: cache db 1, broker db 0. Lượng task lớn thì tách hẳn một Redis riêng cho broker.

Worker nên để bao nhiêu concurrency?

Bắt đầu bằng số lõi CPU, tăng dần theo Flower. Task nặng I/O như gửi email chịu được concurrency cao, task nặng CPU không vượt số lõi.

Task chạy quá lâu thì sao?

task_time_limit kill task cứng sau 300 giây, soft_time_limit cho cơ hội dọn dẹp trước 60 giây. Task hay chạm giới hạn thì chia nhỏ hoặc tối ưu, đừng chỉ tăng giới hạn.

Nguồn tham khảo