Django + Celery 非同步任務佇列入門 | Django 教學

# 安裝 Celery、Redis 客戶端與 Django Celery Results
pip install celery redis django-celery-results

# macOS（使用 Homebrew）
brew install redis
brew services start redis

# Ubuntu / Debian
sudo apt install redis-server
sudo systemctl start redis

# 驗證 Redis 是否正常運作
redis-cli ping
# 預期回應：PONG

# settings.py

# Celery Broker 設定（使用 Redis）
CELERY_BROKER_URL = 'redis://localhost:6379/0'

# Result Backend 設定（使用 Redis）
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'
# 或使用 Django ORM 儲存結果：
# CELERY_RESULT_BACKEND = 'django-db'
# 需在 INSTALLED_APPS 加入 'django_celery_results'

# 序列化設定（使用 JSON 格式）
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

# 時區設定
CELERY_TIMEZONE = 'Asia/Taipei'
CELERY_ENABLE_UTC = True

# 任務結果過期時間（秒）
CELERY_RESULT_EXPIRES = 60 * 60 * 24  # 24 小時後自動清除

# 任務超時設定
CELERY_TASK_SOFT_TIME_LIMIT = 300  # 5 分鐘，拋出 SoftTimeLimitExceeded（可捕捉）
CELERY_TASK_TIME_LIMIT = 600       # 10 分鐘，強制終止（無法捕捉）

# settings.py
INSTALLED_APPS = [
    # ... 其他 App
    'django_celery_results',
]

python manage.py migrate django_celery_results

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

# 設定 Django settings 模組，讓 Celery 能讀取 Django 設定
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

# 建立 Celery App 實例
app = Celery('myproject')

# 從 settings.py 中讀取所有 CELERY_ 開頭的設定
app.config_from_object('django.conf:settings', namespace='CELERY')

# 自動探索所有 INSTALLED_APPS 中的 tasks.py 檔案
app.autodiscover_tasks()


@app.task(bind=True, ignore_result=True)
def debug_task(self):
    """除錯用任務，印出 request 資訊"""
    print(f'Request: {self.request!r}')

# myproject/__init__.py
from .celery import app as celery_app

__all__ = ('celery_app',)

# apps/users/tasks.py
from celery import shared_task
from celery.utils.log import get_task_logger
from django.core.mail import send_mail

logger = get_task_logger(__name__)


@shared_task
def send_welcome_email(user_id: int) -> dict:
    """發送歡迎 Email（最簡單的任務定義）"""
    from apps.users.models import User  # 延遲匯入，避免循環 import

    user = User.objects.get(pk=user_id)
    send_mail(
        subject='歡迎加入！',
        message=f'親愛的 {user.username}，歡迎加入我們的平台。',
        from_email='noreply@example.com',
        recipient_list=[user.email],
    )
    logger.info(f'歡迎信已發送給 user_id={user_id}')
    return {'status': 'success', 'user_id': user_id}

# apps/users/tasks.py
@shared_task(
    bind=True,              # self 參數可存取 task 實例
    max_retries=3,          # 最多重試 3 次
    default_retry_delay=60, # 每次重試間隔 60 秒
)
def send_welcome_email_with_retry(self, user_id: int) -> dict:
    """發送歡迎 Email，失敗時自動重試"""
    from apps.users.models import User

    try:
        user = User.objects.get(pk=user_id)
        send_mail(
            subject='歡迎加入！',
            message=f'親愛的 {user.username}，歡迎加入我們的平台。',
            from_email='noreply@example.com',
            recipient_list=[user.email],
        )
        logger.info(f'歡迎信已發送給 user_id={user_id}')
        return {'status': 'success', 'user_id': user_id}

    except User.DoesNotExist:
        logger.error(f'找不到 user_id={user_id}')
        raise  # 不重試，直接標記失敗

    except Exception as exc:
        logger.warning(f'發送失敗，準備重試：{exc}')
        raise self.retry(exc=exc)  # 觸發重試機制

# views.py
from apps.users.tasks import send_welcome_email

def register_view(request):
    # ... 處理註冊邏輯 ...
    user = User.objects.create_user(username='test', email='test@example.com')

    # 非同步發送歡迎信（立即返回，不等待 Email 發送完成）
    send_welcome_email.delay(user.id)

    return JsonResponse({'message': '註冊成功，歡迎信將稍後寄出'})

from datetime import datetime

# 10 秒後才執行
send_welcome_email.apply_async(
    args=[user.id],
    countdown=10,
)

# 指定執行時間
send_welcome_email.apply_async(
    args=[user.id],
    eta=datetime(2026, 6, 22, 9, 0),
)

# 指定佇列
send_welcome_email.apply_async(
    args=[user.id],
    queue='high_priority',
)

# 設定重試策略
send_welcome_email.apply_async(
    args=[user.id],
    retry=True,
    retry_policy={
        'max_retries': 5,
        'interval_start': 0,    # 第一次重試的等待秒數
        'interval_step': 0.2,   # 每次重試遞增的秒數
        'interval_max': 0.5,    # 最大等待秒數
    },
)

from celery import signature

# 建立任務簽名
task_sig = send_welcome_email.s(user.id)

# 延遲執行
task_sig.delay()

# 也可以直接用 apply_async
task_sig.apply_async(countdown=30)

from apps.users.tasks import send_welcome_email

# 呼叫任務，取得 AsyncResult
result = send_welcome_email.delay(user.id)

# 取得任務 ID（唯一識別碼）
print(result.id)       # 例如：'a1b2c3d4-e5f6-7890-abcd-ef1234567890'

# 檢查任務狀態
print(result.status)   # 'PENDING' / 'STARTED' / 'SUCCESS' / 'FAILURE' / 'RETRY'

# 檢查任務是否完成
print(result.ready())  # True / False

# 檢查任務是否成功
print(result.successful())  # True / False

# 取得任務結果（會阻塞直到任務完成，有 timeout 參數）
print(result.get(timeout=10))  # {'status': 'success', 'user_id': 1}

# 如果任務失敗，取得例外資訊
if result.failed():
    print(result.traceback)

from celery.result import AsyncResult

# 用任務 ID 重新建立 AsyncResult 物件
result = AsyncResult('a1b2c3d4-e5f6-7890-abcd-ef1234567890')
print(result.status)
print(result.result)

# apps/reports/tasks.py
from celery import shared_task
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)


@shared_task(bind=True)
def generate_report(self, report_id: int) -> dict:
    """生成報表（耗時操作）"""
    from apps.reports.models import Report

    report = Report.objects.get(pk=report_id)
    report.status = 'processing'
    report.task_id = self.request.id  # 儲存任務 ID，方便前端查詢進度
    report.save()

    try:
        # 模擬耗時的報表生成邏輯
        data = report.collect_data()
        file_path = report.render_to_pdf(data)

        report.status = 'completed'
        report.file_path = file_path
        report.save()

        logger.info(f'報表 {report_id} 生成完成：{file_path}')
        return {'status': 'completed', 'file_path': file_path}

    except Exception as exc:
        report.status = 'failed'
        report.save()
        logger.error(f'報表 {report_id} 生成失敗：{exc}')
        raise

# apps/reports/views.py
from django.http import JsonResponse
from apps.reports.models import Report
from apps.reports.tasks import generate_report
from celery.result import AsyncResult


def create_report_view(request):
    """建立報表並觸發非同步生成"""
    report = Report.objects.create(
        user=request.user,
        report_type=request.POST['type'],
        status='pending',
    )

    # 觸發非同步任務
    result = generate_report.delay(report.id)

    return JsonResponse({
        'message': '報表生成中，請稍候',
        'report_id': report.id,
        'task_id': result.id,
    })


def check_report_status_view(request, task_id):
    """查詢報表生成進度"""
    result = AsyncResult(task_id)

    return JsonResponse({
        'task_id': task_id,
        'status': result.status,
        'result': result.result if result.ready() else None,
    })

# 啟動 Worker（基本指令）
celery -A myproject worker --loglevel=info

# 指定併發數量（預設為 CPU 核心數）
celery -A myproject worker --concurrency=4 --loglevel=info

# 指定監聽的佇列
celery -A myproject worker -Q default,high_priority --loglevel=info

# 開發環境：使用 solo 模式（單執行緒，方便除錯）
celery -A myproject worker --pool=solo --loglevel=debug