Django中celery的使用项目实例

作者:宠乖仪 时间:2021-01-07 15:53:14 

1、django应用Celery

django框架请求/响应的过程是同步的,框架本身无法实现异步响应。

但是我们在项目过程中会经常会遇到一些耗时的任务, 比如:发送邮件、发送短信、大数据统计等等,这些操作耗时长,同步执行对用户体验非常不友好,那么在这种情况下就需要实现异步执行。

异步执行前端一般使用ajax,后端使用Celery。

2 、项目应用

django项目应用celery,主要有两种任务方式,一是异步任务(发布者任务),一般是web请求,二是定时任务。

celery组成

Celery是由Python开发、简单、灵活、可靠的分布式任务队列,是一个处理异步任务的框架,其本质是生产者消费者模型,生产者发送任务到消息队列,消费者负责处理任务。Celery侧重于实时操作,但对调度支持也很好,其每天可以处理数以百万计的任务。特点:

简单:熟悉celery的工作流程后,配置使用简单

高可用:当任务执行失败或执行过程中发生连接中断,celery会自动尝试重新执行任务

快速:一个单进程的celery每分钟可处理上百万个任务

灵活:几乎celery的各个组件都可以被扩展及自定制

Celery由三部分构成:

消息中间件(Broker):官方提供了很多备选方案,支持RabbitMQ、Redis、Amazon SQS、MongoDB、Memcached 等,官方推荐RabbitMQ

任务执行单元(Worker):任务执行单元,负责从消息队列中取出任务执行,它可以启动一个或者多个,也可以启动在不同的机器节点,这就是其实现分布式的核心

结果存储(Backend):官方提供了诸多的存储方式支持:RabbitMQ、 Redis、Memcached,SQLAlchemy, Django ORM、Apache Cassandra、Elasticsearch等

架构如下:

Django中celery的使用项目实例

工作原理:

任务模块Task包含异步任务和定时任务。其中,异步任务通常在业务逻辑中被触发并发往消息队列,而定时任务由Celery Beat进程周期性地将任务发往消息队列;

任务执行单元Worker实时监视消息队列获取队列中的任务执行;

Woker执行完任务后将结果保存在Backend中;

本文使用的是redis数据库作为消息中间件和结果存储数据库

1.异步任务redis

1.安装库

pip install celery
pip install redis

2.celery.py

在主项目目录下,新建 celery.py 文件:

import os
import django
from celery import Celery
from django.conf import settings

# 设置系统环境变量,安装django,必须设置,否则在启动celery时会报错
# celery_study 是当前项目名
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celery_study.settings')
django.setup()

celery_app = Celery('celery_study')
celery_app.config_from_object('django.conf:settings')
celery_app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

Django中celery的使用项目实例

注意:是和settings.py文件同目录,一定不能建立在项目根目录,不然会引起 celery 这个模块名的命名冲突  

同时,在主项目的init.py中,添加如下代码:

from .celery import celery_app

__all__ = ['celery_app']

Django中celery的使用项目实例

3.settings.py

在配置文件中配置对应的redis配置:

# Broker配置,使用Redis作为消息中间件
BROKER_URL = 'redis://127.0.0.1:6379/0'

# BACKEND配置,这里使用redis
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'

# 结果序列化方案
CELERY_RESULT_SERIALIZER = 'json'

# 任务结果过期时间,秒
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24

# 时区配置
CELERY_TIMEZONE='Asia/Shanghai'  

# 指定导入的任务模块,可以指定多个
#CELERY_IMPORTS = (    
#    'other_dir.tasks',
#)

Django中celery的使用项目实例

注意:所有配置的官方文档:Configuration and defaults — Celery 5.2.0b3 documentation  

4.tasks.py

在子应用下建立各自对应的任务文件tasks.py(必须是tasks.py这个名字,不允许修改)

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)

Django中celery的使用项目实例

5.调用任务

from .tasks import *
# Create your views here.

def task_add_view(request):
   add.delay(100,200)
   return HttpResponse(f'调用函数结果')

Django中celery的使用项目实例

Django中celery的使用项目实例

6.启动celery

pip install eventlet
celery  -A celery_study worker  -l debug -P eventlet

注意 :celery_study是项目名

使用redis时,有可能会出现如下类似的异常

AttributeError: 'str' object has no attribute 'items'

这是由于版本差异,需要卸载已经安装的python环境中的 redis 库,重新指定安装特定版本(celery4.x以下适用 redis2.10.6, celery4.3以上使用redis3.2.0以上):

xxxxxxxxxx pip install redis==2.10.6

7.获取任务结果

在 views.py 中,通过 AsyncResult.get() 获取结果

from celery import result
def get_result_by_taskid(request):
   task_id = request.GET.get('task_id')
# 异步执行
   ar = result.AsyncResult(task_id)

if ar.ready():
       return JsonResponse({'status': ar.state, 'result': ar.get()})
   else:
       return JsonResponse({'status': ar.state, 'result': ''})

Django中celery的使用项目实例

AsyncResult类的常用的属性和方法:

  • state: 返回任务状态,等同status;

  • task_id: 返回任务id;

  • result: 返回任务结果,同get()方法;

  • ready(): 判断任务是否执行以及有结果,有结果为True,否则False;

  • info(): 获取任务信息,默认为结果;

  • wait(t): 等待t秒后获取结果,若任务执行完毕,则不等待直接获取结果,若任务在执行中,则wait期间一直阻塞,直到超时报错;

  • successful(): 判断任务是否成功,成功为True,否则为False;

2.定时任务

在第一步的异步任务的基础上,进行部分修改即可

1.settings.py

from celery.schedules import crontab

CELERYBEAT_SCHEDULE = {
   'mul_every_30_seconds': {
        # 任务路径
       'task': 'celery_app.tasks.mul',
        # 每30秒执行一次
       'schedule': 5,
       'args': (14, 5)
   }
}

Django中celery的使用项目实例

说明(更多内容见文档:Periodic Tasks — Celery 5.2.0b3 documentation):

  • task:任务函数

  • schedule:执行频率,可以是整型(秒数),也可以是timedelta对象,也可以是crontab对象,也可以是自定义类(继承celery.schedules.schedule)

  • args:位置参数,列表或元组

  • kwargs:关键字参数,字典

  • options:可选参数,字典,任何 apply_async() 支持的参数

  • relative:默认是False,取相对于beat的开始时间;设置为True,则取设置的timedelta时间

在task.py中设置了日志

from celery import shared_task
import logging  
logger = logging.getLogger(__name__))

@shared_task
def mul(x, y):
   logger.info('___mul__'*10)
   return x * y

2.启动celery

(两个cmd)分别启动worker和beat

celery -A worker celery_study -l debug -P eventlet
celery beat -A celery_study -l debug

3.任务绑定

Celery可通过task绑定到实例获取到task的上下文,这样我们可以在task运行时候获取到task的状态,记录相关日志等

方法:

  • 在装饰器中加入参数 bind=True

  • 在task函数中的第一个参数设置为self

在task.py 里面写

from celery import shared_task
import logging  
logger = logging.getLogger(__name__)

# 任务绑定
@shared_task(bind=True)
def add(self,x, y):
   logger.info('add__-----'*10)
   logger.info('name:',self.name)
   logger.info('dir(self)',dir(self))
   return x + y

其中:self对象是celery.app.task.Task的实例,可以用于实现重试等多种功能

from celery import shared_task
import logging  
logger = logging.getLogger(__name__)

# 任务绑定
@shared_task(bind=True)
def add(self,x, y):
   try:
       logger.info('add__-----'*10)
       logger.info('name:',self.name)
       logger.info('dir(self)',dir(self))
       raise Exception
   except Exception as e:
       # 出错每4秒尝试一次,总共尝试4次
       self.retry(exc=e, countdown=4, max_retries=4)    
   return x + y

启动celery

celery -A worker celery_study -l debug -P eventlet

4.任务钩子

Celery在执行任务时,提供了钩子方法用于在任务执行完成时候进行对应的操作,在Task源码中提供了很多状态钩子函数如:on_success(成功后执行)、on_failure(失败时候执行)、on_retry(任务重试时候执行)、after_return(任务返回时候执行)

方法:通过继承Task类,重写对应方法即可,

from celery import Task

class MyHookTask(Task):

def on_success(self, retval, task_id, args, kwargs):
       logger.info(f'task id:{task_id} , arg:{args} , successful !')

def on_failure(self, exc, task_id, args, kwargs, einfo):
       logger.info(f'task id:{task_id} , arg:{args} , failed ! erros: {exc}')

def on_retry(self, exc, task_id, args, kwargs, einfo):
       logger.info(f'task id:{task_id} , arg:{args} , retry !  erros: {exc}')

# 在对应的task函数的装饰器中,通过 base=MyHookTask 指定
@shared_task(base=MyHookTask, bind=True)
def add(self,x, y):
   logger.info('add__-----'*10)
   logger.info('name:',self.name)
   logger.info('dir(self)',dir(self))
   return x + y

启动celery

celery -A worker celery_study -l debug -P eventlet

5.任务编排

在很多情况下,一个任务需要由多个子任务或者一个任务需要很多步骤才能完成,Celery也能实现这样的任务,完成这类型的任务通过以下模块完成:

  • group: 并行调度任务

  • chain: 链式任务调度

  • chord: 类似group,但分header和body2个部分,header可以是一个group任务,执行完成后调用body的任务

  • map: 映射调度,通过输入多个入参来多次调度同一个任务

  • starmap: 类似map,入参类似*args

  • chunks: 将任务按照一定数量进行分组

文档:Next Steps — Celery 5.2.0b3 documentation

1.group

urls.py:

path('primitive/', views.test_primitive),

views.py:

from .tasks import *
from celery import group

def test_primitive(request):
   # 创建10个并列的任务
   lazy_group = group(add.s(i, i) for i in range(10))
   promise = lazy_group()
   result = promise.get()
   return JsonResponse({'function': 'test_primitive', 'result': result})

说明:

通过task函数的 s 方法传入参数,启动任务

上面这种方法需要进行等待,如果依然想实现异步的方式,那么就必须在tasks.py中新建一个task方法,调用group,示例如下:

tasks.py:

@shared_task
def group_task(num):
   return group(add.s(i, i) for i in range(num))().get()

urls.py:

path('first_group/', views.first_group),

views.py:

def first_group(request):
   ar = tasks.group_task.delay(10)

return HttpResponse('返回first_group任务,task_id:' + ar.task_id)

2.chain

默认上一个任务的结果作为下一个任务的第一个参数

def test_primitive(request):
   # 等同调用  mul(add(add(2, 2), 5), 8)
   promise = chain(tasks.add.s(2, 2), tasks.add.s(5), tasks.mul.s(8))()
   #  72
   result = promise.get()  
   return JsonResponse({'function': 'test_primitive', 'result': result})

3.chord

任务分割,分为header和body两部分,hearder任务执行完在执行body,其中hearder返回结果作为参数传递给body

def test_primitive(request):
   # header:  [3, 12]
   # body: xsum([3, 12])
   promise = chord(header=[tasks.add.s(1,2),tasks.mul.s(3,4)],body=tasks.xsum.s())()
   result = promise.get()
   return JsonResponse({'function': 'test_primitive', 'result': result})

6、celery管理和监控

celery通过flower组件实现管理和监控功能 ,flower组件不仅仅提供监控功能,还提供HTTP API可实现对woker和task的管理

官网:flower · PyPI

文档:Flower - Celery monitoring tool — Flower 1.0.1 documentation

安装flower

pip install flower

启动flower

flower -A celery_study --port=5555

说明:

  • -A:项目名

  • --port: 端口号

访问

在浏览器输入:http://127.0.0.1:5555

通过api操作

curl http://127.0.0.1:5555/api/workers

来源:https://blog.csdn.net/qq_53582111/article/details/120207740

标签:django,celery,定时任务
0
投稿

猜你喜欢

  • IE7下 filter:Alpha(opacity=xx) 的小问题

    2008-12-02 16:24:00
  • PHP 实现一种多文件上传的方法

    2024-05-03 15:07:11
  • python中利用Future对象异步返回结果示例代码

    2021-09-10 06:59:52
  • 一文看懂JSONP原理和应用

    2024-04-23 09:10:47
  • 简单谈谈Python中的模块导入

    2021-02-24 20:28:17
  • Web 前端优化最佳实践之 Cookie 篇

    2008-06-25 13:00:00
  • Linux下安装Memcached服务器和客户端与PHP使用示例

    2023-10-05 04:32:44
  • vue项目中安装less依赖的过程

    2024-05-10 14:21:12
  • Mysql limit 优化,百万至千万级快速分页 复合索引的引用并应用于轻量级框架

    2024-01-14 00:42:39
  • Refactoring HTML 书评

    2008-07-10 12:00:00
  • Python入门基础之import机制

    2023-12-25 11:39:12
  • 如何用mysql自带的定时器定时执行sql(每天0点执行与间隔分/时执行)

    2024-01-16 01:58:31
  • mysql中各种常见join连表查询实例总结

    2024-01-19 05:31:52
  • 为什么要进行CSS缩写?

    2007-10-29 12:56:00
  • Python反爬机制-验证码功能的具体实现过程

    2023-02-05 18:53:19
  • 详解python播放音频的三种方法

    2021-12-16 09:19:11
  • Python读取YUV文件,并显示的方法

    2023-03-19 11:44:13
  • 通过案例解析python鸭子类型相关原理

    2022-05-03 12:38:57
  • MAC下mysql安装配置方法图文教程

    2024-01-17 22:38:46
  • sql server常用命令行操作(启动、停止、暂停)

    2012-01-05 19:02:48
  • asp之家 网络编程 m.aspxhome.com