Celery使用异步、定时任务使用

一、什么是Celery

1.1、celery是什么

Celery是一个简单、灵活且可靠的,处理大量消息的分布式系统,专注于实时处理的异步任务队列,同时也支持任务调度。

Celery的架构由三部分组成,消息中间件(message broker),任务执行单元(worker)和任务执行结果存储(task result store)组成。

消息中间件

Celery本身不提供消息服务,但是可以方便的和第三方提供的消息中间件集成。包括,RabbitMQ, Redis等等

任务执行单元

Worker是Celery提供的任务执行的单元,worker并发的运行在分布式的系统节点中。

任务结果存储

Task result store用来存储Worker执行的任务的结果,Celery支持以不同方式存储任务的结果,包括AMQP, redis等

另外, Celery还支持不同的并发和序列化的手段

  • 并发:Prefork, Eventlet, gevent, threads/single threaded
  • 序列化:pickle, json, yaml, msgpack. zlib, bzip2 compression, Cryptographic message signing 等等

1.2、使用场景

celery是一个强大的 分布式任务队列的异步处理框架,它可以让任务的执行完全脱离主程序,甚至可以被分配到其他主机上运行。我们通常使用它来实现异步任务(async task)和定时任务(crontab)。

异步任务:将耗时操作任务提交给Celery去异步执行,比如发送短信/邮件、消息推送、音视频处理等等

定时任务:定时执行某件事情,比如每天数据统计

1.3、Celery具有以下优点

Simple(简单)
Celery 使用和维护都非常简单,并且不需要配置文件。

Highly Available(高可用)
woker和client会在网络连接丢失或者失败时,自动进行重试。并且有的brokers 也支持“双主”或者“主/从”的方式实现高可用。

Fast(快速)
单个的Celery进程每分钟可以处理百万级的任务,并且只需要毫秒级的往返延迟(使用 RabbitMQ, librabbitmq, 和优化设置时)

Flexible(灵活)
Celery几乎每个部分都可以扩展使用,自定义池实现、序列化、压缩方案、日志记录、调度器、消费者、生产者、broker传输等等。

 1.4、Celery安装

你可以安装Celery通过Python包管理平台(PyPI)或者源码安装
使用pip安装:

1

$ pip install -U Celery

或着:

1

$ sudo easy_install Celery

二、Celery执行异步任务

2.1、基本使用

创建项目celerypro

创建异步任务执行文件celery_task:

import celery
import time
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"  

创建执行任务文件,生产者:produce_task.py:

from celery_task import send_email
result = send_email.delay("yuan")
print(result.id)
result2 = send_email.delay("alex")
print(result2.id)  

注意,异步任务文件命令执行:

1

celery worker -A celery_app_task -l info

创建py文件消费者:result.py,查看任务执行结果,

from celery.result import AsyncResult
from celery_task import cel

async_result=AsyncResult(id="c6ddd5b7-a662-4f0e-93d4-ab69ec2aea5d", 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('任务已经开始被执行')

2.1、多任务结构

 

celery.py:

from celery import Celery

cel = 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.py,task02.py:

#task01
import time
from celery_tasks.celery import cel

@cel.task
def send_email(res):
    time.sleep(5)
    return "完成向%s发送邮件任务"%res



#task02
import time
from celery_tasks.celery import cel
@cel.task
def send_msg(name):
    time.sleep(5)
    return "完成向%s发送短信任务"%name

produce_task.py:

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)

check_result.py:

from celery.result import AsyncResult
from celery_tasks.celery import cel

async_result = AsyncResult(id="562834c6-e4be-46d2-908a-b102adbbf390", app=cel)

if async_result.successful():
    result = async_result.get()
    print(result)
    # result.forget() # 将结果删除,执行完成,结果不会自动删除
    # async.revoke(terminate=True)  # 无论现在是什么时候,都要终止
    # async.revoke(terminate=False) # 如果任务还没有开始执行呢,那么就可以终止。
elif async_result.failed():
    print('执行失败')
elif async_result.status == 'PENDING':
    print('任务等待中被执行')
elif async_result.status == 'RETRY':
    print('任务异常后正在重试')
elif async_result.status == 'STARTED':
    print('任务已经开始被执行')

开启work:celery worker -A celery_task -l info -P eventlet,添加任务(执行produce_task.py),检查任务执行结果(执行check_result.py)

三、Celery执行定时任务

 设定时间让celery执行一个定时任务,produce_task.py:

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)
task_time = utc_ctime + time_delay

# 使用apply_async并设定时间
result = send_email.apply_async(args=["egon"], eta=task_time)
print(result.id)

多任务结构中celery.py修改如下:

from datetime import timedelta
from celery import Celery
from celery.schedules import crontab

cel = Celery('tasks', broker='redis://127.0.0.1:6379/1', backend='redis://127.0.0.1:6379/2', include=[
    'celery_tasks.task01',
    'celery_tasks.task02',
])
cel.conf.timezone = 'Asia/Shanghai'
cel.conf.enable_utc = False

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': ('张三',)
    # },
} 

1

2

3

# 启动 Beat 程序$ celery beat -A proj<br># Celery Beat进程会读取配置文件的内容,周期性的将配置中到期需要执行的任务发送给任务队列

# 之后启动 worker 进程.$ celery -A proj worker -l info 或者$ celery -B -A proj worker -l info

四、Django中使用celery

 项目根目录创建celery包,目录结构如下:

mycelery/
├── config.py
├── __init__.py
├── main.py
└── sms/
    ├── __init__.py
    ├── tasks.py

配置文件config.py:

broker_url = 'redis://127.0.0.1:6379/15'
result_backend = 'redis://127.0.0.1:6379/14'

任务文件tasks.py:

# celery的任务必须写在tasks.py的文件中,别的文件名称不识别!!!
from mycelerys.main import app
import time


import 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"

最后在main.py主程序中对django的配置文件进行加载

# 主程序
import os
from celery import Celery
# 创建celery实例对象
app = Celery("sms")

# 把celery和django进行组合,识别和加载django的配置文件
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celeryPros.settings.dev')

# 通过app对象加载配置
app.config_from_object("mycelerys.config")

# 加载任务
# 参数必须必须是一个列表,里面的每一个任务都是任务的路径名称
# app.autodiscover_tasks(["任务1","任务2"])
app.autodiscover_tasks(["mycelerys.sms",])

# 启动Celery的命令
# 强烈建议切换目录到mycelery根目录下启动
# celery -A mycelery.main worker --loglevel=info

Django视图调用:

from django.shortcuts import render

# Create your views here.


from django.shortcuts import render,HttpResponse
from mycelerys.sms.tasks import send_sms,send_sms2
from datetime import timedelta

from 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=10)
    # task_time = utc_ctime + time_delay
    # result = send_sms.apply_async(["911", ], eta=task_time)
    # print(result.id)

    return HttpResponse('ok')

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

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

谈谈功率IC巨头—士兰微

大家好&#xff0c;我是砖一。 今天给大家分享一下士兰微电子公司&#xff0c;&#xff0c;有做功率元器件&开关电源和IC的朋友可以了解一下&#xff0c;希望对你有用~ 1 公司介绍 士兰微电子成立于1997年&#xff0c;于2003年上市&#xff0c;总部位于杭州&#xff0c;…

雪花飘,购物抛物线,进度条等四个案列,带入走进 CSS transition

前言 今天从四个案例&#xff0c;我们一起走进 CSS Transition。 源码 以及 在线演示地址 源码地址&#xff1a; 四个案例&#xff0c; CSS Transition 源码 在线演示地址&#xff1a;(兼容移动端) 贝塞尔曲线运动进度条雪花飘飘效果购物车抛物线效果 案例演示 内置贝塞…

kafka(五)——消费者流程分析(c++)

概念 ​ 消费者组&#xff08;Consumer Group&#xff09;&#xff1a;由多个consumer组成。消费者组内每个消费者负责消费不同分区的数据&#xff0c;一个分区只能由一个组内消费者消费&#xff1b;消费者组之间互不影响。所有的消费者都属于某个消费者组&#xff0c;即消费者…

猫头虎博主深度探索:Amazon Q——2023 re:Invent 大会的 AI 革新之星

摘要 大家好&#xff0c;我是猫头虎博主&#xff01;今天&#xff0c;我要带大家深入了解2023年 re:Invent 大会上发布的一款革命性产品——Amazon Q。让我们一起探索这个引领未来工作方式的新型工具吧&#xff01; 引言 在2023年的 re:Invent 大会上&#xff0c;亚马逊云科…

RAG 修炼手册|一文讲透 RAG 背后的技术

在之前的文章中《RAG 修炼手册&#xff5c;RAG敲响丧钟&#xff1f;大模型长上下文是否意味着向量检索不再重要》&#xff0c;我们已经介绍过 RAG 对于解决大模型幻觉问题的不可或缺性&#xff0c;也回顾了如何借助向量数据库提升 RAG 实战效果。 今天我们继续剖析 RAG&#xf…

docker-compose 之 OpenGauss

使用 docker 启动高斯数据库的示范脚本如下&#xff1a; docker-compose.yml version: 3.7 services:opengauss:image: enmotech/opengauss:5.1.0container_name: opengaussnetwork_mode: "host"privileged: truevolumes:- ./opengauss:/var/lib/opengaussenvironm…

前端mock数据——使用mockjs进行mock数据

前端mock数据——使用mockjs进行mock数据 一、安装二、mockjs的具体使用 一、安装 首选需要有nodejs环境安装mockjs&#xff1a;npm install mockjs 若出现像上图这样的错&#xff0c;则只需npm install mockjs --legacy-peer-deps即可 src下新建mock文件夹&#xff1a; mo…

微服务-网关

在微服务架构中&#xff0c;每个服务都是一个可以独立开发和运行的组件&#xff0c;而一个完整的微服务架构由一系列独立运行的微服务组成。其中每个服务都只会完成特定领域的功能&#xff0c;比如订单服务提供与订单业务场景有关的功能、商品服务提供商品展示功能等。各个微服…

SpringBoot文件上传--头像上传

目录 1.在配置文件中写好物理路径和访问路径 2.写配置文件 3.页面上传 4.控制层 5.效果 1.在配置文件中写好物理路径和访问路径 &#xff08;自定义&#xff09;file:uploadPath: D:/upload/img/ 物理路径path: /file/** 访问路径 2.写配置文件 package com.example…

BCLinux8U6系统基线加固致无法su的问题分析

本文对BCLinux8U6系统进行基线加固致无法su的问题分析。 一、问题现象 对BCLinux8U6系统进行基线加固&#xff0c;su切换用户失败&#xff0c;报错信息如下&#xff1a; [ABCDlocalhost ~]$ su - 密码&#xff1a; su: 模块未知 二、问题分析 1、错误排查 出错前&#xf…

探索进程控制第一弹(进程终止、进程等待)

文章目录 进程创建初识fork函数fork函数返回值fork常规用法fork调用失败的原因 写时拷贝进程终止进程终止是在做什么&#xff1f;进程终止的情况代码跑完&#xff0c;结果正确/不正确代码异常终止 如何终止 进程等待概述进程等待方法wait方法waitpid 进程创建 初识fork函数 在…

自定义校验(这里是Validation)

1.自定义注解State package com.itheima.anno;import com.itheima.Validator.StateValidator; import jakarta.validation.Constraint; import jakarta.validation.Payload;import java.lang.annotation.*;Documented//元注解 Target(ElementType.FIELD) Retention(RetentionP…

力扣HOT100 - 239. 滑动窗口最大值

解题思路&#xff1a; class Solution {public int[] maxSlidingWindow(int[] nums, int k) {if(nums.length 0 || k 0) return new int[0];Deque<Integer> deque new LinkedList<>();int[] res new int[nums.length - k 1];// 未形成窗口for(int i 0; i <…

瑞山集团携手数环通iPaaS,实现ERP与CRM无缝对接

01 客户背景 瑞山集团位于粤港澳大湾区核心商圈----佛山市&#xff0c;是一家致力于各种新型材料添加剂应用推广&#xff0c;集科研、销售和服务于一身的新材料企业&#xff0c;产品覆盖了塑料黏胶剂、水性涂料、3D打印材料、电子材料、纳米材料等各种行业。 公司旗下拥有几间不…

pdffactory pro 8注册码序列号下载 附教程

PdfFactory Pro可以说是一款行业专业且技术领先的的PDF虚拟打印机软件。其不仅占用系统内存小巧&#xff0c;功能强大&#xff0c;可支持用户无需使用Acrobat来创建Adobe PDF即可以进行PDF组件的创建和打印。同时&#xff0c;现在全新的PdfFactory Pro 8也正式上线来袭&#xf…

雄安建博会:中矿雄安新区的总部开工建设

中矿落位雄安&#xff1a;助力国家战略与新区发展 雄安新区&#xff0c;作为中国未来发展的重要战略支点&#xff0c;正迎来一系列央企总部的疏解与建设。最近&#xff0c;中国矿产资源集团有限公司&#xff08;简称“中矿”&#xff09;在雄安新区的总部项目正式开工建设&…

在win10上虚拟一个LoongOS系统(类似虚拟机)作为开发环境

文章目录 1.安装1.1.下载这三个东西1.2.安装好qemu。1.3.创建一个启动脚本startup_mate.bat&#xff0c;然后把三部分东西放到一起1.4.然后双击startup.bat就可以启动了。 2.文件的传输2.1.使能虚拟机系统的ssh2.2.连接ssh 3.Qt相关安装Qt安装opencv 1.安装 注意&#xff0c;一…

Web前端—属性描述符

属性描述符 假设有一个对象obj var obj {a:1 }观察这个对象&#xff0c;我们如何来描述属性a&#xff1a; 值为1可以重写可以遍历 我们可以通过Object.getOwnPropertyDescriptor得到它的属性描述符 var desc Object.getOwnPropertyDescriptor(obj, a); console.log(desc);我…

uniapp 2.0可视化开发工具:提升跨平台应用开发效率的新篇章

摘要 随着移动互联网的迅猛发展&#xff0c;跨平台应用开发成为前端开发者关注的热点。uniapp作为一款优秀的跨平台应用框架&#xff0c;其2.0版本的发布为开发者带来了更多的便利和可能性。其中&#xff0c;可视化开发工具的出现更是为前端开发带来了革命性的变革&#xff0c…

Windows下docker-compose部署DolphinScheduler

参照&#xff1a;快速上手 - Docker部署(Docker) - 《Apache DolphinScheduler v3.1.0 使用手册》 - 书栈网 BookStack 下载源文件 地址&#xff1a;https://dolphinscheduler.apache.org/zh-cn/download/3.2.1 解压到指定目录&#xff0c;进入apache-dolphinscheduler-xxx-…