Django 配置 Celery

Django 配置 Celery

celery介绍

Celery Beat 定时任务

异步任务通常在业务逻辑中被触发并发往任务队列，而定时任务由 Celery Beat 进程周期性地将任务发往任务队列

Celery Broker 消息中间件

Celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成。包括RabbitMQ、 Redis等

Celery Worker 任务执行单元

Worker是Celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中

Celery Backend 结果存储

Backend用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括mongodb、redis等

定时任务和异步任务会发送给消息中间件

Celery Worker 会监控中间件消息

Celery Worker 会把执行结果存储到 Backend

安装celery

xxxxxxxxxx
pip install celery

配置Django

Borker 和Backend 都使用redis，可以根据实际进行修改

xxxxxxxxxx
# Celery settings

CELERY_BROKER_URL = 'redis://127.0.0.1/5'
CELERY_ACCEPT_CONTENT = ['json']

CELERY_RESULT_BACKEND = 'redis://127.0.0.1/7'
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai'

application目录下创建tasks.py 用于定义任务

project目录下创建celery.py

project目录下修改__init__.py

xxxxxxxxxx
taskproj
├── taskapp
│   ├── __init__.py
│   ├── apps.py
│   ├── migrations
│   │   └── __init__.py
│   ├── models.py
│   ├── tasks.py
│   └── views.py
├── manage.py
├── taskproj
│   ├── __init__.py
│   ├── celery.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
└── templates

tasks.py

xxxxxxxxxx
from __future__ import absolute_import, unicode_literals
from celery import shared_task


@shared_task
def testAsyncNumAdd(x, y):
    return x + y

@shared_task
def testNum():
    num = 100
    return num

 

celery.py

xxxxxxxxxx
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from celery.schedules import crontab,timedelta

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

app = Celery('taskproj')
# this ‘demo’ is your project name !!!

# 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')

# 定时任务
app.conf.beat_schedule ={
        'testNum':{                                 #取个名字
            'task':'taskapp.tasks.testNum',         #设置是要将哪个任务进行定时
            'schedule': timedelta(hours=24),        #调用crontab 或者timedelta 进行 时间的定义
        },
}

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

__init__.py

xxxxxxxxxx
from __future__ import absolute_import, unicode_literals

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app

__all__ = ('celery_app',)

 

启动celery

启动woker，目前只支持异步任务

xxxxxxxxxx
celery -A taskproj worker -l info

启动 Beat 定时任务调度器

xxxxxxxxxx
celery -A taskproj beat -l info

二合一启动方式，同时支持异步和定时任务

xxxxxxxxxx
celery -A taskproj worker -B -l info

 

备注

由于celery时区默认使用UTC，所以定时任务使用crontab模式时，会出现异常问题

如果非必须指定日期定时任务，建议使用timedelta模式做定时任务