celery的介绍和使用

celery介绍

celery是一个异步任务框架,它可以执行异步任务、延迟任务、定时任务

异步任务框架简述:

1)celery可以不依赖任何服务器,通过自身命令,启动服务(内部支持socket)
2)celery服务为为其他项目服务提供异步解决任务需求的
注:会有两个服务同时运行,一个是项目服务,一个是celery服务,项目服务将需要异步处理的任务交给celery服务,celery就会在需要时异步完成项目的需求

Celery架构

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

image

消息中间件

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

任务执行单元

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

任务结果存储

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

使用场景

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

延迟执行:解决延迟任务

定时执行:解决周期(周期)任务,比如每天数据统计

Celery的安装配置

pip install celery gevent

windows环境下启动celery

# module为python module名
celery -A <module> worker -l info -P gevent

celery执行异步任务

采用包架构封装(多任务结构)

在项目文件project下新建一个celery包

project
    ├── celery_task  	# celery包
    │   ├── __init__.py # 包文件
    │   ├── celery.py   # celery连接和配置相关文件,且名字必须叫celery.py
    │   └── tasks.py    # 所有任务函数
    ├── add_task.py  	# 添加任务
    └── get_result.py   # 获取结果

celery.py

# 导入模块
from celery import Celery
# 指定任务中间件和任务处理仓库为redis里的第一个库,后面1表示第一个,2表示第二个。
broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
# backend='redis://:123456@127.0.0.1:6379/1' redis加密码写法
# include列表里表示celery_task包里的task.py交给app处理
app = Celery(__name__,broker=broker, backend=backend, include=['celery_task.tasks'])

task.py

from .celery import app
import time
@app.task #添加该装饰器表示这个函数任务使用了celery框架的异步处理
def add(n, m):
    print(n)
    print(m)
    time.sleep(10)
    print('n+m的结果:%s' % (n + m))
    return n + m

@app.task
def low(n, m):
    print(n)
    print(m)
    print('n-m的结果:%s' % (n - m))
    return n - m

add_task.py
这个py文件是用来添加任务的,可以建在任意位置

from celery_task import tasks

# 添加立即执行任务
t1 = tasks.add.delay(10, 20)
t2 = tasks.low.delay(100, 50)
print(t1) # 获取的就是这个任务的id号


# 添加延迟任务
from datetime import datetime, timedelta
eta=datetime.utcnow() + timedelta(seconds=10)
tasks.low.apply_async(args=(200, 50), eta=eta)

get_result.py
这个py文件是用来获取任务处理的结果的

from celery_task.celery import app

from celery.result import AsyncResult

id = '21325a40-9d32-44b5-a701-9a31cc3c74b5' #输入任务的id号
if __name__ == '__main__':
    async = AsyncResult(id=id, app=app)
    if async.successful(): #如果这个任务执行成功
        result = async.get() #得到结果
        print(result)
    elif async.failed():
        print('任务失败')
    elif async.status == 'PENDING':
        print('任务等待中被执行')
    elif async.status == 'RETRY':
        print('任务异常后正在重试')
    elif async.status == 'STARTED':
        print('任务已经开始被执行')

celery执行延迟任务

在add_task.py里书写:

# 添加延迟任务
from datetime import datetime, timedelta
# datetime.utcnow()表示获取当前的utc时间,timedelta(seconds=10)表示加10秒timedelta括号里之后是时间类型,可以相加减
eta=datetime.utcnow() + timedelta(seconds=10)
# 使用的是apply_async这个方法,tasks.add这个是函数名,args括号里是函数add需要传的参数,eta必须是utc时间
tasks.add.apply_async(args=(200, 50), eta=eta)

celery执行定时任务

将celery.py添加以下定时配置

# 导入模块
from celery import Celery
# 指定任务中间件和任务处理仓库为redis里的第一个库,后面1表示第一个,2表示第二个。
broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
# backend='redis://:123456@127.0.0.1:6379/1' redis加密码写法
# include列表里表示celery_task包里的task.py交给app处理
app = Celery(__name__,broker=broker, backend=backend, include=['celery_task.tasks'])


# 定时任务配置
# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False

# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
    'add-task': {
        'task': 'celery_task.tasks.add',  # celery_task.tasks.low这个是你函数任务的位置,task是固定的
        'schedule': timedelta(seconds=3), # 每隔三秒执行一次
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点
        'args': (300, 150), # 函数任务需要传入的参数
    }
}

然后在终端下cd到scripts文件夹

执行celery -A celery_task beat
表示开始定时发布任务

celery在django中的使用

需要在celery.py中添加以下代码:

# 加载django环境
import os
import django
# luffyapi.settings.dev 这个是项目的配置文件位置
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "luffyapi.settings.dev")
django.setup()

本站文章如无特殊说明,均为本站原创,如若转载,请注明出处:celery的介绍和使用 - Python技术站

(0)
上一篇 2023年4月2日 下午4:30
下一篇 2023年4月2日

相关文章

  • 微信小程序的点击事件、页面跳转、数据绑定、数据修改、获取用户信息及定位、for指令、获取用户上传图片

    小程序中实现页面跳转 对标签绑定点击事件 data是点击时传入的参数 <view bindtap=”clickMe” data-nid=”123″ data-name=”SD” >点我跳转</view> /** * 用户点击事件 */ clickMe(e){ console.log(e) var nid = e.currentTarg…

    Python开发 2023年4月2日
    00
  • xadmin的使用

    安装 在项目的虚拟环境下执行 pip install https://codeload.github.com/sshwsfc/xadmin/zip/django2 注意:xadmin对于不同django版本有不同的版本,一定要使用相对应的版本 在app中注册 INSTALLED_APPS = [ # … # xadmin主体模块 ‘xadmin’, # …

    Python开发 2023年4月2日
    00
  • 个人站点页面搭建和侧边栏展示功能

    展示个人站点页面 前端代码:和首页文章展示基本一样: <div class=”col-md-8″> {% for article in article_list %} <ul class=”media-list”> <li class=”media”> <a href=””><h4 class=”medi…

    Python开发 2023年4月2日
    00
  • 小程序开发环境搭建

    如何开发微信小程序 小程序:学习微信开发的语言(前端html、css、js、vue.js) 微信开发者工具 API:restful接口(Python+django+drf框架)。 pycharm 环境搭建 python环境搭建 虚拟环境 django drf pycharm 小程序环境 在微信公众平台上注册一个小程序账号 在开发设置里保存好自己的appid …

    2023年4月2日
    00
  • django中有关ajax的部分

    Django_ajax 1 简介 AJAX(Asynchronous Javascript And XML)翻译成中文就是“异步Javascript和XML”。即使用Javascript语言与服务器进行异步交互,传输的数据为XML(当然,传输的数据不只是XML)。 同步交互:客户端发出一个请求后,需要等待服务器响应结束后,才能发出第二个请求; 异步交互:客户…

    Python开发 2023年4月2日
    00
  • 基于django中间件的编程思想

    基于django中间件的编程思想 在django中,中间件的形式全部写成了列表套字符串的形式 MIDDLEWARE = [ ‘django.middleware.security.SecurityMiddleware’, ‘django.contrib.sessions.middleware.SessionMiddleware’, ‘django.middl…

    2023年4月2日
    00
  • 登录功能

    思路分析 登录页面,我们还是采用ajax的方式提交用户数据 唯一需要学习的是如何制作图片验证码! 具体的登录页面效果图如下: 如何制作图片验证码 推导步骤1:在img标签的src属性里放上验证码的请求路径 补充1.img的src属性: 1.图片路径 2.url 3.图片的二进制数据 补充2:字体样式 我们计算机上之所以可以输出各种各样的字体样式,其内部其实对…

    2023年4月2日
    00
  • flask基础介绍和语法

    flask基础介绍和语法 flask项目初始化的app.py介绍 # 从flask包中导入flask类 from flask import Flask # 使用flask类实例化出一个app对象 # __name__:代表当前app.py这个模块 app = Flask(__name__) # 创建一个路由和视图的映射 @app.route(‘/’) def…

    2023年4月2日
    00
合作推广
合作推广
分享本页
返回顶部