# 創(chuàng)建虛擬環(huán)境
python3 -m venv myenv
source myenv/bin/activate

# 安裝依賴
pip install django celery redis django-celery-beat

# 創(chuàng)建項(xiàng)目
django-admin startproject task_manager
cd task_manager

# 創(chuàng)建應(yīng)用
python manage.py startapp tasks

INSTALLED_APPS = [
    # ...
    'django_celery_beat',
    'django_celery_results',
    'tasks',
]

# 數(shù)據(jù)庫配置
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.sqlite3',
        'NAME': BASE_DIR / 'db.sqlite3',
    }
}

# Celery配置
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'django-db'  # 使用django-celery-results存儲(chǔ)結(jié)果
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai'

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'task_manager.settings')

app = Celery('task_manager')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

@app.task(bind=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

from __future__ import absolute_import, unicode_literals
from .celery import app as celery_app

__all__ = ('celery_app',)

from django.db import models
from django.utils import timezone

class ScheduledTask(models.Model):
    TASK_TYPES = (
        ('periodic', '周期性任務(wù)'),
        ('one_time', '一次性任務(wù)'),
    )
    
    name = models.CharField('任務(wù)名稱', max_length=100)
    task_type = models.CharField('任務(wù)類型', max_length=20, choices=TASK_TYPES)
    task_function = models.CharField('任務(wù)函數(shù)', max_length=200)
    cron_expression = models.CharField('Cron表達(dá)式', max_length=100, blank=True, null=True)
    interval_seconds = models.IntegerField('間隔秒數(shù)', blank=True, null=True)
    next_run_time = models.DateTimeField('下次執(zhí)行時(shí)間', blank=True, null=True)
    is_active = models.BooleanField('是否激活', default=True)
    created_at = models.DateTimeField('創(chuàng)建時(shí)間', auto_now_add=True)
    updated_at = models.DateTimeField('更新時(shí)間', auto_now=True)
    
    def __str__(self):
        return self.name
    
    class Meta:
        verbose_name = '定時(shí)任務(wù)'
        verbose_name_plural = '定時(shí)任務(wù)列表'

class TaskExecutionLog(models.Model):
    task = models.ForeignKey(ScheduledTask, on_delete=models.CASCADE, related_name='logs')
    execution_time = models.DateTimeField('執(zhí)行時(shí)間', auto_now_add=True)
    status = models.CharField('執(zhí)行狀態(tài)', max_length=20, choices=(
        ('success', '成功'),
        ('failed', '失敗'),
    ))
    result = models.TextField('執(zhí)行結(jié)果', blank=True, null=True)
    error_message = models.TextField('錯(cuò)誤信息', blank=True, null=True)
    
    def __str__(self):
        return f"{self.task.name} - {self.execution_time}"
    
    class Meta:
        verbose_name = '任務(wù)執(zhí)行日志'
        verbose_name_plural = '任務(wù)執(zhí)行日志列表'

python manage.py makemigrations
python manage.py migrate

from rest_framework import serializers
from .models import ScheduledTask, TaskExecutionLog

class ScheduledTaskSerializer(serializers.ModelSerializer):
    class Meta:
        model = ScheduledTask
        fields = '__all__'

class TaskExecutionLogSerializer(serializers.ModelSerializer):
    class Meta:
        model = TaskExecutionLog
        fields = '__all__'

from rest_framework import viewsets, status
from rest_framework.response import Response
from .models import ScheduledTask, TaskExecutionLog
from .serializers import ScheduledTaskSerializer, TaskExecutionLogSerializer
from celery import current_app
from django_celery_beat.models import PeriodicTask, IntervalSchedule, CrontabSchedule
import json

class ScheduledTaskViewSet(viewsets.ModelViewSet):
    queryset = ScheduledTask.objects.all()
    serializer_class = ScheduledTaskSerializer
    
    def create(self, request, *args, **kwargs):
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        
        # 創(chuàng)建Celery定時(shí)任務(wù)
        task = serializer.save()
        self._create_celery_task(task)
        
        headers = self.get_success_headers(serializer.data)
        return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)
    
    def update(self, request, *args, **kwargs):
        partial = kwargs.pop('partial', False)
        instance = self.get_object()
        serializer = self.get_serializer(instance, data=request.data, partial=partial)
        serializer.is_valid(raise_exception=True)
        
        # 更新Celery定時(shí)任務(wù)
        task = serializer.save()
        self._update_celery_task(task)
        
        return Response(serializer.data)
    
    def destroy(self, request, *args, **kwargs):
        instance = self.get_object()
        
        # 刪除Celery定時(shí)任務(wù)
        self._delete_celery_task(instance)
        
        self.perform_destroy(instance)
        return Response(status=status.HTTP_204_NO_CONTENT)
    
    def _create_celery_task(self, task):
        if task.task_type == 'periodic':
            # 創(chuàng)建間隔調(diào)度
            schedule, _ = IntervalSchedule.objects.get_or_create(
                every=task.interval_seconds,
                period=IntervalSchedule.SECONDS,
            )
            PeriodicTask.objects.create(
                interval=schedule,
                name=task.name,
                task=task.task_function,
                enabled=task.is_active,
                args=json.dumps([]),
                kwargs=json.dumps({}),
            )
        elif task.task_type == 'one_time':
            # 一次性任務(wù)使用ETA
            pass
    
    def _update_celery_task(self, task):
        try:
            periodic_task = PeriodicTask.objects.get(name=task.name)
            if task.task_type == 'periodic':
                schedule, _ = IntervalSchedule.objects.get_or_create(
                    every=task.interval_seconds,
                    period=IntervalSchedule.SECONDS,
                )
                periodic_task.interval = schedule
            periodic_task.enabled = task.is_active
            periodic_task.save()
        except PeriodicTask.DoesNotExist:
            self._create_celery_task(task)
    
    def _delete_celery_task(self, task):
        try:
            periodic_task = PeriodicTask.objects.get(name=task.name)
            periodic_task.delete()
        except PeriodicTask.DoesNotExist:
            pass

class TaskExecutionLogViewSet(viewsets.ReadOnlyModelViewSet):
    queryset = TaskExecutionLog.objects.all()
    serializer_class = TaskExecutionLogSerializer

from django.urls import include, path
from rest_framework import routers
from .views import ScheduledTaskViewSet, TaskExecutionLogViewSet

router = routers.DefaultRouter()
router.register(r'tasks', ScheduledTaskViewSet)
router.register(r'logs', TaskExecutionLogViewSet)

urlpatterns = [
    path('', include(router.urls)),
]

from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('api/', include('tasks.urls')),
]

from celery import shared_task
from .models import ScheduledTask, TaskExecutionLog
import logging

logger = logging.getLogger(__name__)

@shared_task(bind=True, autoretry_for=(Exception,), retry_backoff=3, retry_kwargs={'max_retries': 3})
def sample_task(self, task_id):
    try:
        task = ScheduledTask.objects.get(id=task_id)
        
        # 模擬任務(wù)執(zhí)行
        result = f"任務(wù) {task.name} 執(zhí)行成功，時(shí)間：{str(self.request.time_start)}"
        
        # 記錄執(zhí)行日志
        TaskExecutionLog.objects.create(
            task=task,
            status='success',
            result=result
        )
        
        logger.info(f"任務(wù)執(zhí)行成功: {task.name}")
        return result
        
    except Exception as e:
        # 記錄錯(cuò)誤日志
        task = ScheduledTask.objects.get(id=task_id) if ScheduledTask.objects.filter(id=task_id).exists() else None
        if task:
            TaskExecutionLog.objects.create(
                task=task,
                status='failed',
                error_message=str(e)
            )
        logger.error(f"任務(wù)執(zhí)行失敗: {str(e)}")
        raise

redis-server

celery -A task_manager worker --loglevel=info --pool=prefork --concurrency=4

celery -A task_manager beat --loglevel=info --scheduler django_celery_beat.schedulers:DatabaseScheduler

python manage.py runserver

curl -X POST http://localhost:8000/api/tasks/ -d '{
    "name": "示例周期性任務(wù)",
    "task_type": "periodic",
    "task_function": "tasks.tasks.sample_task",
    "interval_seconds": 60,
    "is_active": true
}' -H "Content-Type: application/json"

curl http://localhost:8000/api/tasks/

curl http://localhost:8000/api/logs/

task_manager/
├── task_manager/
│   ├── __init__.py
│   ├── celery.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
├── tasks/
│   ├── migrations/
│   ├── __init__.py
│   ├── admin.py
│   ├── apps.py
│   ├── models.py
│   ├── serializers.py
│   ├── tasks.py
│   ├── urls.py
│   └── views.py
├── manage.py
└── db.sqlite3