[ django中使用celery ]

发表评论
89 views

安装依赖

  1. pip install celery
  2. pip install celery[librabbitmq]
  3. pip install django-celery-results

settings.py文件

CELERY_BROKER_URL = 'amqp://chf:chenhaifeng@47.120.55.99:5672' CELERY_TASK_TRACK_STARTED = True CELERY_TASK_TIME_LIMIT = 30 * 60 CELERY_BROKER_CONNECTION_RETRY_ON_STARTUP = True

主要配置

celery.py

import os

from celery import Celery
from celery.schedules import crontab


# Set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django_celery.settings')

app = Celery('django_celery')

# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django apps.
app.autodiscover_tasks()


@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    # Calls test('hello') every 10 seconds.
    sender.add_periodic_task(10.0, test_b.s('hello'), name='add every 10')

    # Calls test('hello') every 30 seconds.
    # It uses the same signature of previous task, an explicit name is
    # defined to avoid this task replacing the previous one defined.
    sender.add_periodic_task(30.0, test_b.s('hello'), name='add every 30')

    # Calls test('world') every 30 seconds
    sender.add_periodic_task(30.0, test_b.s('world'), expires=10)

    # Executes every Monday morning at 7:30 a.m.
    sender.add_periodic_task(
        crontab(hour=7, minute=30, day_of_week=1),
        test_b.s('Happy Mondays!'),
    )


@app.task
def test_b(arg):
    print(arg)

# @app.task
# def add(x, y):
#     z = x + y
#     print(z)

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

启动定时任务

celery -A proj beat

# 指定文件
celery -A proj beat -s /home/celery/var/run/celerybeat-schedule

在django中配置消息结果表

你可以通过设置 Celery 的 CELERY_RESULT_BACKEND 为 PostgreSQL 数据库（pg数据库），但Celery 官方并不直接支持 PostgreSQL 作为结果后端。
常见做法有两种：

方案一：使用第三方库 django-celery-results

  1. 安装依赖
    bash pip install django-celery-results

  2. 在 Django 配置中添加 app
    python # filepath: g:\python_objects\gns_work\cews\backend\config\settings\base.py INSTALLED_APPS += ["django_celery_results"]

  3. 设置 Celery 结果后端为 Django ORM
    python # filepath: g:\python_objects\gns_work\cews\backend\config\settings\base.py CELERY_RESULT_BACKEND = "django-db"

  4. 迁移数据库
    bash python manage.py migrate django_celery_results

这样 Celery 的任务结果就会自动保存到你的 PostgreSQL 数据库（Django 默认数据库）。

方案二：任务结果自定义保存

如果你想把结果保存到自定义表，可以在任务里自己写保存逻辑。例如：

from backend.cews.models import MyTaskResult

@celery_app.task
def my_task(...):
    result = ... # 你的任务逻辑
    MyTaskResult.objects.create(result=result)
    return result

推荐用方案一，简单且通用。
设置好后，所有 Celery 任务的返回值都会自动存到 pg 数据库的 django_celery_results_taskresult 表里。
如需查询结果，可以直接查这个表或用 Django ORM 查询。

如需详细配置步骤或代码示例，请告知！

总结

  1. 启动的beat是负责生产

  2. 启动的worker是负责消费

  3. 生产者把消息发到消息中间件

  4. 消费者去取