Message Queue với Celery và Redis: Xử lý tác vụ bất đồng bộ

Người dùng click "Đăng ký" và chờ 3 giây để hệ thống gửi email chào mừng — đó là trải nghiệm tệ không cần thiết. Message queue tách biệt tác vụ nặng ra khỏi request cycle, giúp response trở lại ngay lập tức.

Khi Nào Cần Message Queue

Gửi email/SMS
Resize hoặc process ảnh/video
Generate báo cáo PDF
Sync data với third-party API
Scheduled jobs (cron)
Tác vụ có thể fail và cần retry

Cài Đặt Celery Với Django

pip install celery redis

# blog/celery.py
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'blog.settings')
app = Celery('blog')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

# blog/__init__.py
from .celery import app as celery_app
__all__ = ('celery_app',)

# settings.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

Định Nghĩa Task

# posts/tasks.py
from celery import shared_task
from django.core.mail import send_mail

@shared_task(bind=True, max_retries=3)
def send_welcome_email(self, user_id: int):
    try:
        from django.contrib.auth.models import User
        user = User.objects.get(id=user_id)
        send_mail(
            subject='Chào mừng bạn đến với ducdev blog!',
            message=f'Xin chào {user.get_full_name()}, ...',
            from_email='no-reply@ducdev.com',
            recipient_list=[user.email],
        )
    except Exception as exc:
        # Retry sau 60s, tối đa 3 lần
        raise self.retry(exc=exc, countdown=60)

@shared_task
def generate_sitemap():
    from django.test import RequestFactory
    from django.contrib.sitemaps.views import sitemap
    # Generate và cache sitemap
    pass

Gọi Task Từ View

from posts.tasks import send_welcome_email

def register(request):
    # Tạo user
    user = User.objects.create_user(...)
    # Đẩy vào queue — response trả về ngay, không chờ email
    send_welcome_email.delay(user.id)
    return JsonResponse({'status': 'ok'})

Periodic Tasks (Cron)

# settings.py
from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    'update-reading-time': {
        'task': 'posts.tasks.update_all_reading_times',
        'schedule': crontab(hour=3, minute=0),  # 3am hàng ngày
    },
    'cleanup-old-sessions': {
        'task': 'posts.tasks.cleanup_sessions',
        'schedule': crontab(day_of_week='sunday', hour=2),
    },
}

# Chạy beat scheduler
celery -A blog beat -l info

Chạy Worker

# Development
celery -A blog worker -l info

# Production với concurrency
celery -A blog worker -l info --concurrency=4

# Monitor với Flower
pip install flower
celery -A blog flower --port=5555

Quy tắc đơn giản: nếu tác vụ mất hơn 200ms hoặc có thể fail vì lý do ngoài control (network, third-party API), hãy đưa vào queue.

Kết Luận

Celery + Redis là combo phổ biến nhất cho Django. Setup ban đầu mất khoảng 1-2 giờ nhưng mang lại lợi ích lớn: response time nhanh hơn, retry tự động, và có thể scale worker riêng biệt với web server.

Message Queue — Xử Lý Tác Vụ Bất Đồng Bộ Với Celery Và Redis