python中Celery 异步任务队列的高级用法

from celery import shared_task
from celery.exceptions import Retry

@shared_task(bind=True, max_retries=3, default_retry_delay=5)
def fetch_data(self, url):
    try:
        response = requests.get(url)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as exc:
        # 如果失败则重试
        raise self.retry(exc=exc)

from celery import group

group_tasks = group(task1.s(arg1), task2.s(arg2), task3.s(arg3))
result = group_tasks.apply_async()

from celery import chain

workflow = chain(task1.s(arg1) | task2.s(arg2) | task3.s(arg3))
result = workflow.apply_async()

from celery import chord

workflow = chord(
    [task1.s(arg1), task2.s(arg2)],
    callback_task.s()
)
result = workflow.apply_async()

@shared_task(priority=1)
def high_priority_task():
    # 高优先级任务逻辑
    pass

@shared_task(priority=10)
def low_priority_task():
    # 低优先级任务逻辑
    pass

task_queues:
  - name: my_queue
    exchange: my_exchange
    routing_key: my_key
    queue_arguments:
      x-max-priority: 10

from celery import Celery
from celery.schedules import crontab

app = Celery('my_app')

app.conf.beat_schedule = {
    'add-every-10-seconds': {
        'task': 'my_app.tasks.add',
        'schedule': 10.0,
        'args': (16, 16),
    },
    'daily-task': {
        'task': 'my_app.tasks.daily_report',
        'schedule': crontab(hour=7, minute=30),
    },
}

from django_celery_beat.models import PeriodicTask, IntervalSchedule

# 创建新调度
schedule, created = IntervalSchedule.objects.get_or_create(
    every=10,
    period=IntervalSchedule.SECONDS,
)
PeriodicTask.objects.create(
    interval=schedule,
    name='new_task',
    task='my_app.tasks.some_task',
    args=json.dumps([10, 20]),
)

app.conf.result_expires = 3600  # 结果在 1 小时后过期

celery -A my_app purge

pip install flower
celery -A my_app flower