异步框架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.mzph.cn/news/700873.shtml

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

相关文章

Linux的信号

Linux的信号是一种用于进程之间通信的机制。它们用于向进程发送通知,告知进程发生了某种事件或请求进程执行某个操作。信号可以由内核、其他进程或进程自身发送。 信号的作用有以下几个方面: 通知进程某个事件的发生,如进程的终止、挂起、恢…

springboot/ssm来访管理系统Java访客预约来访登记系统web

springboot/ssm来访管理系统Java访客预约来访登记系统web 基于springboot(可改ssm)vue项目 开发语言:Java 框架:springboot/可改ssm vue JDK版本:JDK1.8(或11) 服务器:tomcat 数据库:mysq…

软件实例,物流货运配货单打印模板软件单据打印查询管理系统软件教程,可以同时打印标签或补打

软件实例,物流货运配货单打印模板软件单据打印查询管理系统软件教程,可以同时打印标签或补打 一、前言 以下软件教程以 佳易王物流单打印查询系统V17.1为例说明 软件文件下载可以点击最下方官网卡片——软件下载——试用版软件下载 这个版本在原来基…

Android 跨进程通信技术优劣分析

一、引言 在Android开发中,跨进程通信(IPC)是一项常见的任务,用于在不同的应用程序或组件之间交换数据。有多种IPC机制可供选择,每种机制都有其优势和劣势。以下是对几种常见IPC技术的分析。 二、跨进程通信 2.1、Bi…

【软件使用】postman使用教程

​ 🍎个人博客:个人主页 🏆个人专栏:软件安装及使用 ⛳️ 功不唐捐,玉汝于成 ​ 目录 前言 正文 步骤1:安装Postman 步骤2:发送请求 步骤3:管理环境变量 步骤4&#xff1…

Leetcode 26-30题

删除有序数组中的重复项 给定一个有序数组,要求原地删除重复出现的元素,返回删除后的数组的长度。 这里的原地删除其实可以这样表示,用双指针从前往后扫一遍,遇到新的没出现过的元素就放到前面去,就可以实现删除后的数…

Linux线程同步(2)死锁与互斥锁

死锁(Deadlock)是指两个或两个以上的进程(或线程)在执行过程中,由于竞争资源或者由于彼此通信而造成的一种阻塞的现象,若无外力作用,它们都将无法推进下去。此时称系统处于死锁状态或系统产生了…

定频空调与变频空调的区别

变频空调是通过变频器来改变压缩机运转电压以及频率&#xff0c;从而改变压缩机转速的新型空调&#xff0c;那么他和定频空调有哪些区别呢&#xff1f; 一、控制启停的方式不同 定频空调控制方式为&#xff1a;当T环 < T设 - 2℃&#xff0c;压缩机停机。 当 T环 > T…

Java实现就医保险管理系统 JAVA+Vue+SpringBoot+MySQL

目录 一、摘要1.1 项目介绍1.2 项目录屏 二、功能模块2.1 科室档案模块2.2 医生档案模块2.3 预约挂号模块2.4 我的挂号模块 三、系统展示四、核心代码4.1 用户查询全部医生4.2 新增医生4.3 查询科室4.4 新增号源4.5 预约号源 五、免责说明 一、摘要 1.1 项目介绍 基于JAVAVue…

【前端素材】推荐优质后台管理系统Qovex平台模板(附源码)

一、需求分析 1、定义 后台管理系统是一种用于管理和监控网站、应用程序或系统的在线工具。它通常是通过网页界面进行访问和操作&#xff0c;用于管理网站内容、用户权限、数据分析等。后台管理系统是网站或应用程序的控制中心&#xff0c;管理员可以通过后台系统进行各种管理…

构造函数,原型,实例,类的关系整理

视频来源js原型链、构造函数和类_哔哩哔哩_bilibili 如视频所说&#xff0c;构造函数的prototype指向原型&#xff0c;实例化的对象的__proto__指向原型&#xff0c;原型通过constructor指向构造函数&#xff0c;正如class里面的constructor方法就相当于Person构造函数一样&am…

基于PID-bang-bang控制算法的卫星姿态控制matlab仿真

目录 1.课题概述 2.系统仿真结果 3.核心程序与模型 4.系统原理简介 5.完整工程文件 1.课题概述 基于PID-bang-bang控制算法的卫星姿态控制。仿真输出控制器的控制收敛曲线&#xff0c;卫星姿态调整过程的动画。 2.系统仿真结果 3.核心程序与模型 版本&#xff1a;MATLAB…

连接SFTP服务器报错host key for has changed and you have requested strict checking

背景&#xff1a;今天连接SFTP的时候&#xff0c;报错如下&#xff0c;记录一下 报错如下 WARNING: REMOTE HOST IDENTIFICATION HSS CHANGED!cccececccccceeteeacaaeapepeppeeecpeepeeecepeeeaeeeeeceeeeeeIT IS POSSIBLE THAT SOMEONE IS DOING SOMETHING NASTY!Someone co…

LeetCode每日一题——2583. Kth Largest Sum in a Binary Tree

文章目录 一、题目二、题解 一、题目 You are given the root of a binary tree and a positive integer k. The level sum in the tree is the sum of the values of the nodes that are on the same level. Return the kth largest level sum in the tree (not necessaril…

前后端联调可能出现的问题

调不到后端数据 前后端传参方式不一样 --- formdata 主要接收文件 或者有文件和数据 --- json 纯数据

智能成绩表(C语言)

题目来自于博主算法大师的专栏&#xff1a;最新华为OD机试C卷AB卷OJ&#xff08;CJavaJSPy&#xff09; https://blog.csdn.net/banxia_frontend/category_12225173.html 题目描述 小明来到某学校当老师&#xff0c;需要将学生按考试总分或单科分数进行排名&#xff0c;你能帮…

springboot+vue项目部署配置开机自启动

1.前端部属 下载nginx解压&#xff0c;在nginx\conf下找到nginx.conf 添加如下代码 server {listen 8081;server_name localhost;charset utf-8;location / {root F:/1ceshi/dist; #前端打包路径try_files $uri $uri/ /index.html;index index.html index.htm;}l…

本地创建Git仓库

在 Windows 下&#xff0c;可以通过以下步骤在本地创建一个 并模拟远程Git 仓库。 1、在命令行中打开模拟远程Git 仓库目标文件夹&#xff1a; 打开命令提示符或 PowerShell。例如&#xff1a; 创建裸仓库&#xff08;模拟远程仓库&#xff09;&#xff1a;创建一个裸仓库&am…

Redis进阶篇

Redis线程模型 redis是基于内存运行的高性能k-v数据库&#xff0c;6.x之前是单线程, 对外提供的键值存储服务的主要流程 是单线程&#xff0c;也就是网络 IO 和数据读写是由单个线程来完成&#xff0c;6.x之后引入多线程而键值对读写命 令仍然是单线程处理的&#xff0c;所以 …

【PX4SimulinkGazebo联合仿真】在Simulink中使用ROS2控制无人机进入Offboard模式起飞悬停并在Gazebo中可视化

在Simulink中使用ROS2控制无人机进入Offboard模式起飞悬停并在Gazebo中可视化 系统架构Matlab官方例程Control a Simulated UAV Using ROS 2 and PX4 Bridge运行所需的环境配置PX4&Simulink&Gazebo联合仿真实现方法建立Simulink模型并完成基本配置整体框架各子系统实现…