Django结合Celery进行异步调用

目录

Celery介绍

相关环境

相关配置

1、在proj/proj/目录下创建一个新的celery.py模块

定义 Celery 实例：

2、在proj/proj/__init__.py 模块中导入这个应用程序。

3、在各自模块中定义任务文件tasks.py

4、settings.py配置

 服务启动

 异步调用

 

---

Celery介绍

Celery 是一个简单，灵活且可靠的分布式系统，可以处理大量消息。它是一个任务队列，着重于实时处理，同时还支持任务调度。

Celery 可以做队列、异步调用、解耦、高并发、流量削峰、定时任务等等

一个 Celery 系统可以由多个 worker 和 broker 组成，从而实现高可用性和横向扩展。

Celery既可以独立使用，也可以结合Django使用

原理：celery将待处理任务扔进消息队列，然后由worker进程进行消费

相关环境

Python 3.8
django-celery-beat==2.5.0
django-celery-results==2.5.1
celery==5.2.7
redis==4.6.0

此文中用redis做消息队列，用Mysql5.7保存执行结果

相关配置

 Django 项目布局如下：

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

1、在proj/proj/目录下创建一个新的celery.py模块

定义 Celery 实例：

import os
 
from celery import Celery
 
# Set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')
 
app = Celery('proj')
 
# 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.task(bind=True, ignore_result=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings') 将设置模块传递给设置模块传递给celery程序
app.config_from_object('django.conf:settings', namespace='CELERY')从配置文件中获取以CELERY开头的配置

2、在proj/proj/__init__.py 模块中导入这个应用程序。

主要是为了应用程序在 Django 启动时加载，以便@shared_task装饰器：

proj/proj/__init__.py：

# 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、在各自模块中定义任务文件tasks.py

Celery 会自动从tasks.py发现这些任务

#一个简单的tasks任务
 
from celery import shared_task
 
 
@shared_task
def add(x, y):
    return x + y

目录结构如下

- proj/
  - manage.py
  - proj1/
    - tasks.py
    - admin.py
    - models.py
    - views.py
    - tests.py
    - apps.py
  - proj/
    - __init__.py
    - settings.py
    - celery.py
    - urls.py

4、settings.py配置

CELERY_BROKER_URL= xxxx.xxx.xxx.xxx
# 消息队列地址，可以是redis、mq、mysql等等
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_TIMEZONE = "Asia/Shanghai"
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 30 * 60
CELERY_ENABLE_UTC = False
DJANGO_CELERY_BEAT_TZ_AWARE = False
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'
CELERY_RESULT_BACKEND = 'django-db'
# 任务执行结果保存方式
CELERY_RESULT_BACKEND_TRANSPORT_OPTIONS= {
    'global_keyprefix': '{}_celery_prefix_'.format(ENV_PROFILE)
}
CELERYD_CONCURRENCY = 5
# 执行任务的并发工作进程/线程/线程的数量。
CELERYD_MAX_TASKS_PER_CHILD = 100
# 任务池工作进程在被新进程替换之前可以执行的最大任务数。默认没有限制。
CELERYD_FORCE_EXECV = True

CELERY_TASK_TRACK_STARTED：当True时，任务判断是否将其状态报告为“已启动”。默认值是False。任何长时间运行任务并且需要报告当前正在运行的任务。

CELERY_TASK_TIME_LIMIT：任务时间限制（以秒为单位）。当超过这个值时，处理任务进程将被杀死并被新的进程取代。

CELERY_ACCEPT_CONTENT：收到的消息进行序列化，默认情况下仅启用 json，但可以添加任何内容类型，包括 pickle 和 yaml；

注意：

1、celery从4.0引入了小写配置，但是从 Django 设置模块加载 Celery 配置，需要继续使用大写名称。而且还需要使用CELERY_前缀，以便 Celery 设置不会与其他应用程序使用的 Django 设置发生冲突。

2、用Mysql5.7作为任务结果存储，数据库设置为utf8，如果使用utf8mb4无法成功创建相关表

 服务启动

celery -A proj worker -l INFO
#定时任务启动方式如下：
celery -A itmanager beat 

异步调用

views.py代码如下

#一个简单的tasks任务
 
from proj1.tasks import add
 
 
def add(request):
    x = int(request.POST.get('x', '0'))
    y = int(request.POST.get('y', '0'))
    addInfo.delay(x, y)

默认情况下不启用保存结果。本文我们已经Celery 结果后端。在admin管理后台可以看到任务运行情况

常见问题

1、结果未保存或任务始终处于PENDING状态

所有任务默认为PENDING，因此状态为“未知”。发送任务时，Celery 不会更新状态，并且任何没有历史记录的任务都被假定为待处理。

1. 确保该任务ignore_result未启用。

   启用此选项将强制工作人员跳过更新状态

2. 确保没有任何旧线程在运行。

   很容易意外启动多个工作线程，因此在启动新工作线程之前，请确保前一个工作线程已正确关闭。

   未配置预期结果后端的旧工作人员可能正在运行并劫持任务。

   可以将该参数--pidfile设置为绝对路径以确保不会发生这种情况。

3. 确保客户端配置了正确的后端。

   如果由于某种原因，客户端配置为使用与工作线程不同的后端，您将无法收到结果。确保后端配置正确：

   >>> result = task.delay()
   >>> print(result.backend)