Celery 与 Django

-1.前言

Django原生是单线程的，如果遇到执行时间过长的,只能干等着页面返回，而且不能做别的事情。为了解决这种状况,决定采用异步任务(Celery)的方式。
Celery对Windows的兼容性比较差，建议使用Linux系统。但本篇文章，还是在Windows上测试的，不要问为什么，问就是因为穷。
关于Celery的介绍，我们已经在另一篇文章celery概述介绍过了，这一篇是直接干活的文章。示例来自官方文档，本人稍作注释改动。

环境&版本明细

• Windows操作系统
• Python 3.6.8
• Celery 4.4
• Django 2.2.2

• Redis
  注：请提前安装好Python，Django，和Redis.

  参考文献

• 官方文档

0.目录

• 1.Using Celery with Django
• 2.详解celery.py模块
• 3.使用@shared_task装饰
• 4.扩展
• 5.启动Woreker进程

1.Using Celery with Django

Celery之前需要一个单独的库(djcelery)与Django一起使用，但是从3.1.x(ps:3.1之后就是4.x版本)以后就不再使用了。只需要安装好Celery，Django就可以直接使用了。经过测试发现Celery4.x版本依然可以是和djcelery一起使用,但真的没必要。

要将Celery与Django项目一起使用,必须首先定义Celery库的实例(称为"app")。
如果已经有了一个Django项目,例如:

- proj/
  - manage.py
  - proj/
    - __init__.py
    - settings.py
    - urls.py

那么建议的方法是创建一个新的proj/proj/celery.py模块，该模块定义Celery实例:
文件: proj/proj/celery.py

from __future__ import absolute_import, unicode_literals

import os

from celery import Celery

# set the defalut Django settings module for the 'celery' program
# 为"celery"程序设置默认的Django settings 模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')

app = Celery('proj')

# Using a string here means the worker dosen't have to serialize the configuration object to child processes.
# 在这里使用字符串意味着worker不必将配置对象序列化为子进程。
# - namespace='CELERY' means all celery-related configuration keys should have a 'CELERY_' prefix
# namespace="CELERY"表示所有与Celery相关的配置keys均应该带有'CELERY_'前缀。
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django app configs.
# 从所有注册的Django app 配置中加载 task模块。
app.autodiscover_tasks()


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

然后，需要将此模块proj/proj/celery.py导入proj/proj/init.py中。这样可以确保再Django启动时加载该应用,以便@shared_task装饰器(稍后提及)将使用该应用。
文件：proj/proj/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.
这将确保在Django启动时始终导入应用程序，以便shared_task使用该应用程序。
from .celery import app as celery_app

__all__ = ('celery_app',)

2.详解celery.py模块

接下来分解一下celery.py模块中发生的情况，更好的来理解。
首先我们从future中导入absolute_import,这样celery.py模块就不会与库冲突:

from __future__ import absolute_import

然后,为celery命令行程序设置默认DJANGO_SETTINGS_MODULE的环境变量:

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

您不需要此行，但是可以避免始终将设模块传递到celery程序中。他必须始终在创建应用程序实例之前出现，接下来要做的是:

app = Celery('proj')

这是Celery库实例，可以有很多实例，但是使用Django时真的没必要这么干。
我们还需要将Django settings模块添加为Celery的配置源。这意味着可以不必使用多个配置文件，从而直接从Django settings中配置Celery。也可以根据要将他们分开。。

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

大写命名空间意味着所有Celery配置选项必须以大写而不是小写指定,并以开头CELERY_,例如,task_always_eager设置为CELERY_TASK_ALWAYS_EAGER。这也适用于工作程序设置,例如,worker_concurrency设置为CELERY_WORKER_CONCURRENCY。
可以直接传递设置对象，建议使用字符串，因为那样，worker就不必序列化该对象。该CELERY_命名空间也是可选的，但是建议使用这个，为了更好的识别和防止与其他设置冲突。
在这里我简单的罗列一下我的设置，有啥需求的，自己点击官方文档Celery配置选项去学习。

文件:Django的settings.py

CELERY_BROKER_URL = 'redis://localhost:6379/1'  # Broker配置，使用Redis作为消息中间件

CELERY_RESULT_BACKEND = 'redis://localhost:6379/2'  # Backend设置，使用redis作为后端结果存储

CELERY_TIMEZONE = 'Asia/Shanghai'

CELERY_ENABLE_UTC = False

CELERYD_FORCE_EXECV = True  # 防止任务死锁

CELERYD_CONCURRENCY = 10  # 并发的worker数量

CELERY_ACKS_LATE = True

CELERYD_MAX_TASKS_PER_CHILD = 5  # 每个worker最多执行的任务数, 可防止内存泄漏

CELERYD_TASK_TIME_LIMIT = 15 * 60  # 任务超时时间

接下来，可重用的应用程序单独常见做法是在一个单独的tasks.py模块中定义所有任务，而Celery可以自动发现这些模块:

app.autodiscover_tasks()

使用上边这一行,Celery将按照约定自动发现所有已经安装的app中的tasks.py。

- app1/
    - tasks.py
    - models.py
- app2/
    - tasks.py
    - models.py

这样，就可以不必手动将单个模块添加到CELERY_IMPORTS设置中。

最后，该debug_task示例是一个转储其自己的请求信息的任务。这是使用Celery3.1中引入的新 bind=True任务选项，可以轻松地引用当前任务实例。

3.使用@shared_task装饰器

您编写的任务可能会存在于可重用的app中，并且可重用的apps不能依赖于项目本身，因此您也不能直接导入app实例。
该@shared_task装饰器可以让你无需任何具体的app实例创建任务:

文件:app1/tasks.py

from __future__ import absolute_import, unicode_literals
from celery import shared_task


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

@shared_task
def mul(x, y):
    return x * y


@shared_task
def xsum(numbers):
    return sum(numbers)

调用任务,函数.delay(参数)，示例如下:

from app1.tasks import add
res = add.delay(2, 3)
res.get()

附:官方完整示例代码:https : //github.com/celery/celery/tree/master/examples/django/

4.扩展

django-celery-results

作用：使用Django ORM/Cache作为结果存储。

要将其用于项目，需要执行以下步骤:

• 安装django-celery-results库:

$ pip install django-celery-results

• 添加django_celery_results到Django项目的settins.py文件中的INSTALLED_APPS里:

INSTALLED_APPS = (
    ...,
    'django_celery_results',
)
注意：python模块名称是没有短划线的,只有下划线的。

• 通过执行数据库迁移来创建Celery数据库表:

$ python manage.py migrate celery_results

• 配置Celery以使用django-celery-results后端

# 如果使用的是Django settings.py配置Celery，添加以下设置:
CELERY_RESULT_BACKEND = 'django-db'
# 对于缓存后端,可以使用
CELERY_CACHE_BACKEND = 'django-cache'

django-celery-beat

作用：具有管理界面的数据库支持的定期任务

这个后期单独整理，详情参见官方文档

5.启动worker进程

In a production environment you’ll want to run the worker in the background as a daemon - see Daemonization - but for testing and development it is useful to be able to start a worker instance by using the celery worker manage command, much as you’d use Django’s manage.py runserver

在生产环境中，你会想要在后台作为守护程序运行woerker, 详情参见官方文档系统守护进程
对于测试和开发环境，可以通过使用Celery worker 管理命令启动，就像Django的 manage.py runserver:

# Linux 操作系统
# 首先,启动redis,省略
# 其次,启动Django项目,省略
# 最后,启动celery worker
$ celery -A 项目名称 worker -l info

# Windows 操作系统
redis-server.exe redis.windows.conf
python manage.py runserver
celery -A 项目名称 worker -l info -P eventlet  

更多命令详解，使用help命令：

$ celery help

Celery 与 Django

原文：https://www.cnblogs.com/mr-five/p/12197145.html