首页 > 其他分享> 文章详细

celery基础

2020-07-25 12:03:45 阅读：215 来源： 互联网

标签：Celery worker app 基础 celery 任务执行

一、celery介绍
二、celery架构
三、celery使用

一、celery介绍

Celery 官网：http://www.celeryproject.org/

Celery 官方文档英文版：http://docs.celeryproject.org/en/latest/index.html

Celery 官方文档中文版：http://docs.jinkan.org/docs/celery/

异步任务框架
执行异步任务
执行延迟任务
执行定时任务

二、celery架构

# 如果 Celery对象:Celery(...) 是放在一个模块下的
# 1）终端切换到该模块所在文件夹位置：scripts
# 2）执行启动worker的命令：celery worker -A 模块名 -l info -P eventlet
# 注：windows系统需要eventlet支持，Linux与MacOS直接执行：celery worker -A 模块名 -l info
# 注：模块名随意


# 如果 Celery对象:Celery(...) 是放在一个包下的
# 1）必须在这个包下建一个celery.py的文件，将Celery(...)产生对象的语句放在该文件中
# 2）执行启动worker的命令：celery worker -A 包名 -l info -P eventlet
# 注：windows系统需要eventlet支持，Linux与MacOS直接执行：celery worker -A 模块名 -l info
# 注：包名随意

消息中间件

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

任务执行单元

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

任务结果存储

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

使用场景

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

延迟执行：解决延迟任务

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

三、celery使用

安装：pip install celery

根据celery架构，我们可以看出，worker就像是一个工人，一直在工作，如果有工作的时候就会从broker（Redist）里去拿过来执行，并放到pakend（redist）中存放结果

方式一：在一个文件夹内的三个页面

worker执行页面

import celery

# broker存储的位置
broker = 'redis://127.0.0.1:6379/1'
# backend存储的位置
backend ='redis://127.0.0.1:6379/2'

# 实例化的celery对象
app=celery.Celery(__name__,broker=broker,backend=backend)

# 需要添加的任务
@app.task
def add(x,y):
    print(x*y)
    return x+y

broker提交任务的页面

from celery_test import add
# 执行这个文件，就是把这个任务添加到数据库中，只要worker在工作
# 就会把这个任务从数据库1中拿出来执行，并把结果放到数据库2中
ret = add.delay(3,4)
# ret 是这个任务的uuid，用于获取任务结果

backend获取任务结果的页面

from celery_test import app

from celery.result import AsyncResult
# 任务对象的唯一标识：uuid
id = '19dc2faa-39f9-47b6-af77-e9d3a4d05d2e'
if __name__ == '__main__':
    async1 = AsyncResult(id=id, app=app)
    if async1.successful():
        result = async1.get()
        print(result)
    elif async1.failed():
        print('任务失败')
    elif async1.status == 'PENDING':
        print('任务等待中被执行')
    elif async1.status == 'RETRY':
        print('任务异常后正在重试')
    elif async1.status == 'STARTED':
        print('任务已经开始被执行')

方式二：worker单独做一个项目文件，添加任务和获取结果分离出来（执行异步任务）

创建一个celery项目（包），内部必须含有名字为celery的py文件，在内部创建celery对象

celery.py

import celery


broker = 'redis://127.0.0.1:6379/1'
backend ='redis://127.0.0.1:6379/2'

app=celery.Celery(__name__,broker=broker,backend=backend,include=['celery_task.task1','celery_task.task2'])

编写需要添加的任务也在这个包内，可创建不同的任务文件，可添加多个

task1.py

from .celery import app

@app.task
def add(x,y):
    print(x,y)
    return x+y

执行延迟任务

添加任务页面

# 执行延迟任务就是多个一个时间参数
# 这里注意，时间参数是根据utc时间，并不是中国时间

from datetime import datetime, timedelta
# 时间对象必须和时间对象相加
eta=datetime.utcnow() + timedelta(seconds=10)
add.apply_async(args=(200, 50), eta=eta)

添加定时任务

celery页面

# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False

# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
    'low-task': {
        'task': 'celery_task.tasks.low',
        'schedule': timedelta(seconds=3),
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点
        'args': (300, 150),
    }
}

# 定时任务的添加必须要新启动一个beat命令去工作
# celery beat -A celery_task -l info

标签：Celery,worker,app,基础,celery,任务,执行
来源： https://www.cnblogs.com/hz2lxt/p/13376159.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9