异步框架Celery在Django中的运用

参考博客:https://www.cnblogs.com/pyedu/p/12461819.html

参考视频:01 celery的工作机制_哔哩哔哩_bilibili


定义:简单灵活、处理大量消息的分布式系统,专注于实时处理异步队列,支持任务调度

主要架构:

  1. 消息中间件:message broker 可以集成第三方消息中间件如Redis、RabbitMQ
  2. 任务执行单元:worker 是celery提供的执行的任务执行的单元,并发分布在分布式的系统节点中
  3. 任务执行结果存储:task result store来存储执行任务的结果,支持方式 redis、AMQP

同步请求: 顺序进行IO操作等待阻塞进程依次执行

异步请求:异步进行,当IO操作阻塞时放到执行单元中完成放到数据库中而不影响其他单元的执行,当主进程需要阻塞的进程结果时会向是数据库中取出该数据(即将耗时操作放到异步队列中不影响主进程的执行),继续向下进行

使用场景:

  1. 异步任务:将耗时操作任务提交到celery异步执行,如:发送短信、消息推送、音视频处理
  2. 定时任务:定时执行某件事情,如:每日数据统计

主要优点:

  • 简单:使用和维护不要配置文件,只需添加基本信息的配置
  • 高可用:在work和client网络连接丢失或失败时会自动进行重试
  • 快速:单个celery进程可每分钟处理百万级任务,只需要毫秒级的往返延迟
  • 灵活:可以扩展使用,自定义池的实现、序列化、日志记录、消费者、broker消息传输

安装:

pip install celery

实践案例:

"""
异步任务执行文件:celery_task.py
消费者模型
"""
import celery
import time
# task.py
import osos.environ.setdefault('FORKED_BY_MULTIPROCESSING', '1')backend='redis://127.0.0.1:6379/1'
broker='redis://127.0.0.1:6379/2'
cel=celery.Celery('test',backend=backend,broker=broker)
@cel.task
def send_email(name):print("向%s发送邮件..."%name)time.sleep(5)print("向%s发送邮件完成"%name)return "ok"@cel.task
def send_msg(name):print("向%s发送短信..."%name)time.sleep(5)print("向%s发送短信完成"%name)return "ok"""""
执行任务文件: produce_task.py
生成者模型
"""
from celery_task import send_email,send_msg
result = send_email.delay("yuan") # 当执行delay函数时会自动调用消息中间件的任务执行队列,放到任务执行单元中
print(result.id)
result = send_msg.delay("alex")
print(result.id)

先启动redis进程

 使用特定命令下发指令执行celery任务:

(注意celery5.0之前的命令是不一样的:celery worker -A celery_task -l info)

 先执行produce_task.py

返回ID: 

 fd27bc20-ccac-4855-9b3d-150708bad2a6
c07cb5b1-845a-44c4-963b-7ce3f92b98c8

 检查celery的异步队列查看执行结果

 注:当遇到以下情况

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "D:\python3\lib\site-packages\billiard\pool.py", line 361, in workloop
    result = (True, prepare_result(fun(*args, **kwargs)))
  File "D:\python3\lib\site-packages\celery\app\trace.py", line 664, in fast_trace_task
    tasks, accept, hostname = _loc
ValueError: not enough values to unpack (expected 3, got 0)
[2024-02-24 15:31:20,394: ERROR/MainProcess] Task handler raised error: ValueError('not enough values to unpack (expected 3, got 0)')

解决方法:

在消费者模型中添加以下代码

import os
os.environ.setdefault('FORKED_BY_MULTIPROCESSING', '1')

 查看异步执行的结果:

"""
查看任务执行结果: result.py
"""
from celery.result import AsyncResult
from celery_task import celasync_result=AsyncResult(id="fd27bc20-ccac-4855-9b3d-150708bad2a6", app=cel)if async_result.successful():result = async_result.get()print(result)# result.forget() # 将结果删除
elif async_result.failed():print('执行失败')
elif async_result.status == 'PENDING':print('任务等待中被执行')
elif async_result.status == 'RETRY':print('任务异常后正在重试')
elif async_result.status == 'STARTED':print('任务已经开始被执行')# 运行结果是上面执行返回的结果:
ok 

celery多任务结构下异步执行:注意celery_tasks的celery名字是固定,不然会报错

# celery
from celery import Celerycel = Celery('celery_demo',broker='redis://127.0.0.1:6379/1',backend='redis://127.0.0.1:6379/2',# 包含以下两个任务文件,去相应的py文件中找任务,对多个任务做分类include=['celery_tasks.task01','celery_tasks.task02'])# 时区
cel.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
cel.conf.enable_utc = False# task01
import time
from .celery import cel@cel.task
def send_email(res):time.sleep(5)return "完成向%s发送邮件任务"%res# task02
import time
from .celery import cel
@cel.task
def send_msg(name):time.sleep(5)return "完成向%s发送短信任务"%name# """"
执行任务文件: produce_task.py  和上面的celery_task保持在同一级目录
生成者模型
"""
from celery_tasks.task01 import send_email
from celery_tasks.task02 import send_msg# 立即告知celery去执行test_celery任务,并传入一个参数
result = send_email.delay('yuan')
print(result.id)
result = send_msg.delay('yuan')
print(result.id)
E:\desktop\my_drf\celerypro>celery -A celery_tasks worker -l info -P eventlet

运行结果:

 定时任务的配置:

# 更新produce_task 文件,增加定时任务
from celery_task import send_email
from datetime import datetime# 方式一
# v1 = datetime(2020, 3, 11, 16, 19, 00)
# print(v1)
# v2 = datetime.utcfromtimestamp(v1.timestamp())
# print(v2)
# result = send_email.apply_async(args=["egon",], eta=v2)  #  定时任务
# print(result.id)# 方式二
ctime = datetime.now()
# 默认用utc时间
utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
from datetime import timedelta
time_delay = timedelta(seconds=10)  # 当时时间10s后执行任务
task_time = utc_ctime + time_delay# 使用apply_async并设定时间
result = send_email.apply_async(args=["egon"], eta=task_time)
print(result.id)# 更新setting
cel.conf.beat_schedule = {# 名字随意命名'add-every-10-seconds': {# 执行tasks1下的test_celery函数'task': 'celery_tasks.task01.send_email',# 每隔2秒执行一次# 'schedule': 1.0,# 'schedule': crontab(minute="*/1"),'schedule': timedelta(seconds=6),# 传递参数'args': ('张三',)},# 'add-every-12-seconds': {#     'task': 'celery_tasks.task01.send_email',#     每年4月11号,8点42分执行#     'schedule': crontab(minute=42, hour=8, day_of_month=11, month_of_year=4),#     'args': ('张三',)# },
} 

运行结果:

 根据上述配置每6s执行task01发送邮件任务

注意:

# 周期性执行任务单元,要注意先启动beat进程而后执行worker单元
E:\desktop\my_drf\celerypro>celery -A celery_tasks beat
E:\desktop\my_drf\celerypro>celery -A celery_tasks worker -l info -P eventlet注意: 当打开beat后而若没有打开worker执行单元会导致beat进程不断向数据库中加入数据

 

  查看redis堆积的数据方法:cmd命令如下

 python脚本实现:

 celery结合django中集成的运用

# tasks
# celery的任务必须写在tasks.py的文件中,别的文件名称不识别!!!
from mycelery.main import app
import timeimport logging
log = logging.getLogger("django")@app.task  # name表示设置任务的名称,如果不填写,则默认使用函数名做为任务名
def send_sms(mobile):"""发送短信"""print("向手机号%s发送短信成功!"%mobile)time.sleep(5)return "send_sms OK"@app.task  # name表示设置任务的名称,如果不填写,则默认使用函数名做为任务名
def send_sms2(mobile):print("向手机号%s发送短信成功!" % mobile)time.sleep(5)return "send_sms2 OK"# config
broker_url = 'redis://127.0.0.1:6379/15'
result_backend = 'redis://127.0.0.1:6379/14'# main
# 主程序
import os
from celery import Celery
# 创建celery实例对象
app = Celery("sms")
# import os
os.environ.setdefault('FORKED_BY_MULTIPROCESSING', '1') # 注意: 默认配置要这样配置,下列的配置会找不到组件导致失败
# 把celery和django进行组合,识别和加载django的配置文件
# os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celerypro.settings.dev')
# os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.local")# 通过app对象加载配置
app.config_from_object("mycelery.config")# 加载任务
# 参数必须必须是一个列表,里面的每一个任务都是任务的路径名称
# app.autodiscover_tasks(["任务1","任务2"])
app.autodiscover_tasks(["mycelery.sms",])# view 
from django.shortcuts import render,HttpResponse
from mycelery.sms.tasks import send_sms,send_sms2
from datetime import timedeltafrom datetime import datetime
def test(request):################################# 异步任务# 1. 声明一个和celery一模一样的任务函数,但是我们可以导包来解决send_sms.delay("110")send_sms2.delay("119")# send_sms.delay() #  如果调用的任务函数没有参数,则不需要填写任何内容################################# 定时任务ctime = datetime.now()# 默认用utc时间utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())time_delay = timedelta(seconds=3) # 3s 发送消息task_time = utc_ctime + time_delayresult = send_sms.apply_async(["911", ], eta=task_time)print(result.id)return HttpResponse('ok')

启动Celery的命令

# 强烈建议切换目录到mycelery根目录下启动
# E:\desktop\my_drf\celerypro>celery -A mycelery.main worker --loglevel=info

运行结果:

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/263138.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

5分钟JavaScript快速入门

目录 一.JavaScript基础语法 二.JavaScript的引入方式 三.JavaScript中的数组 四.BOM对象集合 五.DOM对象集合 六.事件监听 使用addEventListener()方法添加事件监听器 使用onX属性直接指定事件处理函数 使用removeEventListener()方法移除事件监听器 一.JavaScript基础…

Linux日志轮替

文章目录 1. 基本介绍2. 日志轮替文件命名3. logrotate 配置文件4. 把自己的日志加入日志轮替5. 日志轮替机制原理6. 查看内存日志 1. 基本介绍 日志轮替就是把旧的日志文件移动并改名,同时建立新的空日志文件,当旧日志文件超出保存的范围之后&#xff…

Git基本指令

从远程拉代码 git clone gitgitlab-internal.wedobest.com.cn:dengyanhui/gittest.git添加所有文件到待上传列表 git add .提交 git commit -m message推送 git push获取现在的状态 git status更新本地代码 git pullgit拉取某一分支代码 git clone -b develop XXX本地删除…

开源博客项目Blog .NET Core源码学习(9:Autofac使用浅析)

开源博客项目Blog使用Autofac注册并管理组件和服务,Autofac是面向.net 的开源IOC容器,支持通过接口、实例、程序集等方式注册组件和服务,同时支持属性注入、方法注入等注入方式。本文学习并记录Blog项目中Autofac的使用方式。   整个Blog解…

视频推拉流EasyDSS视频直播点播平台授权出现激活码无效并报错400是什么原因?

视频推拉流EasyDSS视频直播点播平台集视频直播、点播、转码、管理、录像、检索、时移回看等功能于一体,可提供音视频采集、视频推拉流、播放H.265编码视频、存储、分发等视频能力服务,在应用场景上,平台可以运用在互联网教育、在线课堂、游戏…

CSS学习(三)

目录: 1. CSS引入方式 1.1 三种样式表 1.2 内部样式表(嵌入式引入) 1.3 行内样式表(内联样式表) 1.4 外部样式表 1.5 总结 1. CSS引入方式 1.1 三种样式表 1.2 内部样式表(嵌入式引入) …

7.(数据结构)堆

7.1 相关概念 堆(Heap)在计算机科学中是一种特殊的数据结构,它通常被实现为一个可以看作完全二叉树的数组对象。以下是一些关于堆的基本概念: 数据结构: 堆是一个优先队列的抽象数据类型实现,通过完全二叉树…

unity ui界面优化

优化一个比较复杂的界面,里面有多个rt和组件。 在初次打开这个界面的时候会发生1s多的卡顿,还是非常严重的。 分析 通过profiler分析 1.打开界面时卡顿。 分析:除了update和dotween相关逻辑,主要在于打开时的lua function调用…

svn客户端下载、安装、使用

下载、使用 打开360软件管家,选怎宝库,搜索svn,点击安装 可以修改安装路径 使用 在桌面右键弹出菜单,点击 输入地址,点击ok 输入用户名、密码 ,等待检出完成

【web】nginx+php环境搭建-关键点(简版)

一、nginx和php常用命令 命令功能Nginxphp-fpm启动systemctl start nginxsystemctl start php-fpm停止systemctl stop nginxsystemctl stop php-fpm重启systemctl restart nginxsystemctl restart php-fpm查看启动状态systemctl status nginxsystemctl status php-fpm开机自启…

3ds Max视频怎么渲染 3ds Max云渲染操作

在3ds Max软件中制作视频动画本质上是逐帧生成画面,并将这些连续帧串联起来创造出动态连贯的视觉效果。常见的视频帧率包括25 FPS(每秒帧数)、60 FPS、以及120 FPS等,帧率的提升可以使视频动画更加流畅。在实质上,视频渲染就是动画渲染&#…

uniapp-提现功能(demo)

页面布局 提现页面 有一个输入框 一个提现按钮 一段提现全部的文字 首先用v-model 和data内的数据双向绑定 输入框逻辑分析 输入框的逻辑 为了符合日常输出 所以要对输入框加一些条件限制 因为是提现 所以对输入的字符做筛选,只允许出现小数点和数字 这里用正则实现的 小数点…

Linux--自定义shell

shell shell就是操作系统提供给用户与操作系统进行交互的命令行界面。它可以理解为一个用户与操作系统之间的接口,用户可以通过输入命令来执行各种操作,如文件管理、进程控制、软件安装等。Shell还可以通过脚本编程实现自动化任务。 常见的Unix系统中使…

BERT学习笔记

论文:《BERT: Pre-training of Deep Bidirectional Transformers for Language Understanding》,2019 代码:[tensorflow],[pytorch] 来源:李沐精度BERT 0、摘要 与之前模型的区别: GPT考虑的是一个单向…

RabbitMQ 部署方式选择

部署模式 RabbitMQ支持多种部署模式,可以根据应用的需求和规模选择适合的模式。以下是一些常见的RabbitMQ部署模式: 单节点模式: 最简单的部署方式,所有的RabbitMQ组件(消息存储、交换机、队列等)都运行在…

【Java程序设计】【C00278】基于Springboot的数码论坛管理系统(有论文)

基于Springboot的数码论坛管理系统(有论文) 项目简介项目获取开发环境项目技术运行截图 项目简介 这是一个基于Springboot的数码论坛系统 本系统分为系统功能模块、管理员功能模块以及用户功能模块。 系统功能模块:在系统首页可以查看首页、…

Linux Android USB gadget(从设备驱动)

Linux Android USB gadget 一:Linux usb gadget 与 Android Composite Gadget二:原生方式和Android方式如何配置函数调用逻辑内核配置原生驱动android驱动三:mass_storage配置虚拟化U盘四:遍历usb设备五:adb usb判断usb设备为adb获取adb配置信息adb设备序列号发送与接收《Linux…

maven的聚合和生命周期

什么是maven的聚合呢?就是父类直接将子类项目一起统一打包安装统一maven的生命周期 1.maven的生命周期 2.在父亲类pom文件指定需要打包的项目 实例代码: <!--maven的聚合 通过modules指定需要打包的maven项目--> <modules><module>../ithema-jopo</m…

【Java】RestClient的使用

RestClient的使用 先导入Maven坐标&#xff0c;要和elasticsearch和kibana的版本保持一致 <dependency><groupId>org.elasticsearch.client</groupId><artifactId>elasticsearch-rest-high-level-client</artifactId><version>7.12.1<…

神经网络系列---计算图基本原理

文章目录 计算图符号微分符号微分的步骤示例符号微分在计算图中的使用总结 数值微分前向差分法中心差分法数值微分的使用注意事项总结 自动微分1. 基本原理2. 主要类型3. 计算图4. 应用5. 工具和库6. 优点和缺点 计算图1. **计算图的建立**2. **前向传播**3. **反向传播**4. **…