【异步/定时任务】Django中使用Celery实现异步和定时任务【原创】

文章目录

参考

官方文档Celery在Django上
官方文档Celery
Celery中文文档
Github地址
Celery全面学习笔记
分布式任务队列Celery入门与进阶

一、概要

Celery是由Python开发、简单、灵活、可靠的分布式任务队列,其本质是生产者消费者模型,生产者发送任务到消息队列,消费者负责处理任务。

Celery侧重于实时操作,但对调度支持也很好,其每天可以处理数以百万计的任务。特点:

  • 简单:熟悉celery的工作流程后,配置使用简单
  • 高可用:当任务执行失败或执行过程中发生连接中断,celery会自动尝试重新执行任务
  • 快速:一个单进程的celery每分钟可处理上百万个任务
  • 灵活:几乎celery的各个组件都可以被扩展及自定制

应用场景举例:

  • web应用:当用户在网站进行某个操作需要很长时间完成时,我们可以将这种操作交给Celery执行,直接返回给用户,等到Celery执行完成以后通知用户,大大提好网站的并发以及用户的体验感。比如发送短信/邮件、消息推送、音视频处理等
  • 任务场景:比如在运维场景下需要批量在几百台机器执行某些命令或者任务,此时Celery可以轻松搞定。
  • 定时任务:向定时导数据报表、定时发送通知类似场景,虽然Linux的计划任务可以帮我实现,但是非常不利于管理,而Celery可以提供管理接口和丰富的API。

Celery由以下四部分构成:任务模块(Task)、消息中间件(Broker)、任务执行单元Worker、结果存储(Backend)

  • 任务模块Task:包含异步任务和定时任务,异步任务通常在业务逻辑中被触发并发往任务队列,而定时任务由Celery Beat进程周期性的将任务发往任务队列
  • 消息中间件Broker:即为任务调度队列,接收任务生产者发来的消息(即任务),将任务存入队列。Celery本身不提供队列服务,官方提供了很多备选方案,支持RabbitMQ、Redis、Amazon SQS等,官方推荐RabbitMQ。
  • 任务执行单元Worker:Worker是任务执行单元,负责从消息队列中取出任务执行,它可以启动一个或者多个,也可以启动在不同的机器节点,这就是其实现分布式的核心。实时监控消息队列,获取队列中调度的任务并执行它。
  • 结果存储Backend:用于存储任务的执行结果。官方提供了诸多的存储方式支持:RabbitMQ、 Redis、Memcached、SQLAlchemy、Django ORM、Apache Cassandra、Elasticsearch

如图:

《【异步/定时任务】Django中使用Celery实现异步和定时任务【原创】》

工作原理:

  • 任务模块Task包含异步任务和定时任务。其中,异步任务通常在业务逻辑中被触发并发往消息队列,而定时任务由Celery Beat进程周期性地将任务发往消息队列;
  • 任务执行单元Worker实时监视消息队列获取队列中的任务执行;
  • Woker执行完任务后将结果保存在Backend中;

如图:

《【异步/定时任务】Django中使用Celery实现异步和定时任务【原创】》

二、环境

  • Python:3.7
  • Django:2.2.4
  • Celery:4.3

注意:Celery 4.x和3.x版本相差很大,建议使用4.x,另外,如果是Django 1.8之前的版本,只能使用Celery 3.x

三、依赖

需要安装的依赖有:

pip install celery
pip install redis
# 或者是直接:pip install "celery[redis]"

pip install mysqlclient
pip install django-celery-results

以上的依赖中,只有Celery是必须的,其他的依赖都不是必须,如下:

  • redis:非必须,默认Celery是使用RabbitMQ作为Broker中间人的并且默认安装好RabbitMQ依赖了,我自己习惯用Redis,所以需要安装Redis依赖
  • mysqlclient:非必须,默认Celery是不保存任务结果的,如果想要查看任务的结果并且可以保存到数据库中,所以必须安装mysqlclient连接数据库
  • django-celery-results:非必须,默认Celery是不保存任务结果的,如果想要查看任务的结果并且保存到数据库中,就必须安装该依赖,如果不想保存到数据库的话,也可以使用Redis来进行保存

中间人:

  • RabbitMQ:默认
  • Redis
  • Amazon SQS

注意:Celery之前的版本是支持以数据库作为中间人的,但是4.x不支持数据库作为中间人,具体可以参考 官方文档

注意:有一些教程使用django-celery这个库来安装celery,这个是Celery3.x的版本,4.x已经不建议使用这个了

四、初始化

1. 新建一个应用

新建一个应用,应用名随意,这里的应用名为tasks

python manage startapp tasks

setting.py中增加应用:

INSTALLED_APPS = (
    ...,
    'tasks',
    'django_celery_results',
)

执行迁移

python manage.py migrate django_celery_results

2. setting.py同级目录修改init.py

setting.py同级目录修改__init__.py:

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

3. setting.py同级目录新建celery.py

setting.py同级目录新建celery.py:

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from django.conf import settings

# 获取当前文件夹名,即为该Django的项目名
project_name = os.path.split(os.path.abspath('.'))[-1]
project_settings = '%s.settings' % project_name

# 设置环境变量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', project_settings)

# 实例化Celery
app = Celery(project_name)

# 使用Django的settings文件配置celery
app.config_from_object('django.conf:settings', namespace='CELERY')

# Celery加载所有注册的应用
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)


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

注意:设置环境变量的时候

os.environ.setdefault('DJANGO_SETTINGS_MODULE', project_settings)

project_settings一定是项目名.settings,如果项目名和目录名不一样的话,则不能使用上面的方式来获取当前文件夹名字了

4. 修改setting.py

setting.py中最下面新增:

# Broker的地址
CELERY_BROKER_URL = 'redis://127.0.0.1:6379'

# 任务执行返回结果
CELERY_RESULT_BACKEND = 'django-db'

# celery内容等消息的格式设置
CELERY_ACCEPT_CONTENT = ['application/json', ]
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

# celery时区设置,使用settings中TIME_ZONE同样的时区
CELERY_TIMEZONE = TIME_ZONE

更多的配置可以参考 官方文档

Redis中间人的配置格式为:

redis://:password@hostname:port/db_number

默认使用的是 localhost 的 6379 端口中 0 数据库

Celery默认是不保存任务的结果,但是支持保存任务的功能,内置一些任务的后端结果存储:

  • 数据库:如果想要使用数据库的话,需要安装django-celery-results,并且CELERY_RESULT_BACKEND需要设置为:django-db,默认存在 django_celery_results_taskresult 表里面
  • Memcached
  • RPC(RabbitMQ/AMQP)
  • Redis:格式跟Redis中间人的配置格式一致,使用Redis的话不需要安装django-celery-results

五、测试

1. 新增任务

在tasks应用下新增tasks.py:

from __future__ import absolute_import, unicode_literals
from celery import shared_task
import datetime

@shared_task
def add(x, y):
    res = x + y
    time_format = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print('当前时间为:' + time_format + ' ,两个数相加的结果为:')
    print(res)
    return res

@shared_task
def mul(x, y):
    res = x * y
    time_format = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print('当前时间为:' + time_format + ' ,两个数相乘的结果为:')
    print(res)
    return res

@shared_task
def xsum(numbers):
    res = sum(numbers)
    print(res)
    return res

最终的目录结构如下:

项目名
│
├─manage.py
│
├─tasks
│ │ admin.py
│ │ apps.py
│ │ models.py
│ │ tests.py
│ │ views.py
| | tasks.py
│ └ __init__.py
│
├─templates
│
└─项目名
    │ settings.py
    │ urls.py
    │ wsgi.py
    | celery.py
    └ __init__.py

2. 启动Worker

在命令行下执行:

celery -A 项目名 worker -l info
# 或者是
celery -A 项目名 worker --loglevel=info

注意:如果是在Windows 10下,并且是Celery 4.x,请使用下面的命令启动,否则在测试的时候会报错,具体见最下面的问题集。

celery -A 项目名 worker --pool=solo -l info

3. 测试

创建一个异步任务来测试,在命令行下执行:

Django2Demo>python manage.py shell
Python 3.7.0b4 (v3.7.0b4:eb96c37699, May  2 2018, 19:02:22) [MSC v.1913 64 bit (AMD64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
(InteractiveConsole)
>>> from app.tasks.tasks import add, mul, xsum
>>> res = add.delay(2,3)
>>> res
<AsyncResult: cc215e80-240d-47a2-8987-5e7367e185a2>
>>> res.get()
5
>>> res.id
'cc215e80-240d-47a2-8987-5e7367e185a2'
>>> res.ready()
True

看看有没有结果输出,或者是可以在Admin管理页面的Celery Results里面看到结果

delay() 是 apply_async() 的快捷方法,即执行异步任务,异步任务由Worker来进行处理,可以通过控制台输出的日志进行查看执行情况。
调用任务会返回一个 AsyncResult 的实例,用于检测任务的状态,等待任务完成获取返回值(如果任务执行失败,会抛出异常)。

其他的命令还有:

from tasks.tasks import add, mul, xsum

# 调用异步任务
t = add.delay(2, 3)

# 同步拿结果
t.get()

# 同步拿结果,并且设置获取结果的超时时间,超过1秒拿不到结果则报错
t.get(timeout=1)

# 检查任务是否完成
t.ready()

# 如果出错,获取错误结果,不触发异常
t.get(propagate=False)
t.traceback

更多的命令可以查看 官方文档

六、定时任务

1. 概要

在Django中实现定时任务,基本上有两套方案:

  • Celery原生:比较简单,在配置中使用CELERY_BEAT_SCHEDULE来人为加载定时任务
  • 后台管理任务:稍微复杂一点,安装django-celery-beat,在Admin后台中管理和配置定时任务

注意:Celery 4.x和3.x的定时任务不兼容,建议使用4.x

2.Celery原生

使用Celery原生方案很简单, 不需要安装其他的依赖,只需要在配置中配置CELERY_BEAT_SCHEDULE即可

配置

在setting.py最后面加上:

from celery.schedules import crontab
from datetime import timedelta

# 定时任务
CELERY_BEAT_SCHEDULE = { 
    # 每十秒执行一次add方法
    'add-every-10-seconds': { 
        'task': 'tasks.tasks.add',

        # 多长时间执行一次
        'schedule': 10.0, # 支持直接用数字表示秒数
        # 'schedule': timedelta(seconds=10), # 可以用timedelta对象

        # 必要的参数,这里指add()的参数
        'args': (16, 26)
    },
    # 每个周一的20:57分执行一次mul方法
    'add-every-monday-morning': { 
        'task': 'tasks.tasks.mul',
        'schedule': crontab(hour=20, minute=57, day_of_week=1),
        'args': (16, 16),
    },
}

更多的用法可以参考 官方文档

启动

命令行上执行:

# 启动worker
celery -A 项目名 worker --pool=solo -l info

# 启动定时任务beat
celery -A 项目名 beat -l info

即可在命令行上看到输出结果,或者是可以在Admin管理页面的Celery Results里面看到结果

《【异步/定时任务】Django中使用Celery实现异步和定时任务【原创】》

3. 后台管理任务

使用后台管理任务的话,可以更方便的使用Django Admin进行所有定时任务的管理以及查看每次任务的结果,而且把任务存到Django的数据库里面去,不过相比起原生方案,步骤比较多且麻烦,而且需要安装django-celery-beat,django-celery-beat会在admin后台生成定时任务模块进行管理,模块为:PERIODIC TASKS,一般为Periodic tasks,保存在django_celery_beat_periodictask 表

配置

安装django-celery-beat

pip install django-celery-beat

setting.py中增加应用:

INSTALLED_APPS = (
    ...,
    'django_celery_beat',
)

执行迁移,会自动创建一些表:

python manage.py migrate django_celery_beat

登录Django Admin后,可以看到多了PERIODIC TASKS模块,在Periodic tasks里面配置定时任务即可,配置定时任务很简单,比如:

《【异步/定时任务】Django中使用Celery实现异步和定时任务【原创】》

《【异步/定时任务】Django中使用Celery实现异步和定时任务【原创】》

《【异步/定时任务】Django中使用Celery实现异步和定时任务【原创】》

启动

命令行上执行:

# 启动worker
celery -A 项目名 worker --pool=solo -l info

# 启动定时任务beat
celery -A 项目名 beat -l info -S django
# 或者是
celery -A 项目名 beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler

即可在命令行上看到输出结果,或者是可以在Admin管理页面的Celery Results里面看到结果,并且启动之后,会在根目录下生成celerybeat.pid文件

注意:使用django-celery-beat来管理后台任务的时候,启动beat需要在后面加上-S参数来指定调度器

注意:这两个方案也可以同时使用,启动的时候指定-S参数即可同时使用的,如果一起使用的话,原生设置的定时任务,会被同步到Django Admin后台里面的定时任务去

如果想要省事的话,也可以把–scheduler django_celery_beat.schedulers:DatabaseScheduler写到配置文件setting.py中去,即

# 定时任务来源 从数据库中读取
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

然后启动定时任务的时候,就可以省略后面了,即:

celery -A 项目名 beat -l info

注意:如果setting.py中的CELERY_TIMEZONE设置的是非UTC的话,比如说’Asia/Shanghai’,且USE_TZ设置为False的话,那么运行定时任务的时候会报错,这个时候需要在配置中加上以下设置可以避免报错,具体的报错可以参考最下面的问题集。

# 避免时区的问题
CELERY_ENABLE_UTC = False
DJANGO_CELERY_BEAT_TZ_AWARE = False

注意:worker和beat也可以同时启动:

celery -B -A 项目名 -l info

七、问题集

1. 报错:ValueError(‘not enough values to unpack (expected 3, got 0)’)

参考:
https://blog.csdn.net/qq_30242609/article/details/79047660
https://segmentfault.com/q/1010000010560344
https://github.com/celery/celery/issues/4178

在Windows平台下,并且是Celery 4.x,在命令行上执行后Worker可能会有报错,报错如下:

Task handler raised error: ValueError('not enough values to unpack(expected 3, got 0)')
Traceback (most recent call last):
  File "C:\Users\boli.hong\AppData\Roaming\Python\Python37\site-packages\billiard\pool.py", line 362, in workloop
    result = (True, prepare_result(fun(*args, **kwargs)))
  File "C:\Users\boli.hong\AppData\Roaming\Python\Python37\site-packages\celery\app\trace.py", line 546, in _fast_trace_task
    tasks, accept, hostname = _loc
ValueError: not enough values to unpack (expected 3, got 0)

原因:
看别人描述大概就是说Celery 4.x不支持Windows平台了,所以Windows下运行就会出现这个问题

解决方法:

  • 第一种方法(建议):–pool=solo,只需要启动Worker的时候,加上参数即可,即celery -A 项目名 worker –pool=solo -l info
  • 第二种方法:eventlet,需要先pip install eventlet,然后启动Worker的时候加上-P参数,celery -A 项目名 worker -l info -P eventlet

2. MySQL backend does not support timezone-aware datetimes when USE_TZ is False.

当setting.py中的CELERY_TIMEZONE设置的是非UTC的话,比如说’Asia/Shanghai’,且USE_TZ设置为False的话,那么启动定时任务的时候会报错,报错信息如下:

 raise ValueError("MySQL backend does not support timezone-aware datetimes when USE_TZ is False.")
ValueError: MySQL backend does not support timezone-aware datetimes when USE_TZ is False.

解决方法:

  • 第一种方法:setting.py中的USE_TZ需要设置为True(不推荐)
  • 第二种方法(推荐):setting.py中增加:
# 避免时区的问题
CELERY_ENABLE_UTC = False
DJANGO_CELERY_BEAT_TZ_AWARE = False

    原文作者:jiandanokok
    原文地址: https://blog.csdn.net/jiandanokok/article/details/102335091
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞