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

celery的使用

2020-07-26 06:31:26 阅读：259 来源： 互联网

标签：Celery task app celery 任务使用 print

Celery

官方

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

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

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

Celery是一个简单、灵活且可靠的，处理大量消息的分布式系统

专注于实时处理的异步任务队列

同时也支持任务调度

注意：

# Celery is a project with minimal funding, so we don’t support Microsoft Windows. Please don’t open any issues related to that platform.

Celery异步任务框架

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

人是一个独立运行的服务 | 医院也是一个独立运行的服务
    正常情况下，人可以完成所有健康情况的动作，不需要医院的参与；但当人生病时，就会被医院接收，解决人生病问题
    人生病的处理方案交给医院来解决，所有人不生病时，医院独立运行，人生病时，医院就来解决人生病的需求
"""

Celery架构

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

消息中间件

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

任务执行单元

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

任务结果存储

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

使用场景

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

延迟执行：解决延迟任务

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

Celery的安装配置

# pip install celery

消息中间件：RabbitMQ/Redis

app=Celery(‘任务名’, broker=’xxx’, backend=’xxx’)

两种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执行异步任务

基本结构

基本结构使用

# 注，流程
# 1）创建app + 任务

# 2）启动celery(app)服务：
    # 非windows
        # 命令：celery worker -A celery_task -l info
    # windows：（需要安装eventlet模块）
        # pip install eventlet
        # celery worker -A celery_task -l info -P eventlet

# 3）添加任务到broker：手动添加，要自定义添加任务的脚本，右键执行脚本

# 4）获取结果：手动获取，要自定义获取任务的脚本，右键执行脚本

celery_task.py

# 创建py文件：celery_task.py
from celery import Celery

broker = 'redis://127.0.0.1:6379/1'  # broker任务队列
backend = 'redis://127.0.0.1:6379/2'  # 结构存储，执行完的结果存在这
# 生成对象
app = Celery(__name__, broker=broker, backend=backend)


# 2 添加任务(使用这个装饰器装饰，@app.task)，app就是上面生成的app对象
@app.task
def add(x, y):
    print(x, y)
    return x + y

add_task.py

# 添加任务到broker
from celery_task import add

# add(3,4)  # 直接执行，不会被添加到broker中
ret = add.delay(5, 4)  # 向broker中添加一个任务，.delay是装饰器给加上的
print(ret.id)

get_result.py

# 查看任务执行结果

from celery_task import app
from celery.result import AsyncResult
id = 'b4d8a28c-d935-476d-a982-0eb55eefe195'
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('任务已经开始被执行')

包架构封装（多任务结构）--推荐的使用方法

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
broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])  # 需要把celery_task包下的tasks任务文件include注册一下

tasks.py

from .celery import app
import time
@app.task
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

from celery_task import tasks

# 添加立即执行任务(异步)
t1 = tasks.add.delay(10, 20)
t2 = tasks.low.delay(100, 50)
print(t1.id)    # 2d4ad592-9548-4c7c-8df4-7f8583e8a1b1


# 添加延迟任务
from datetime import datetime, timedelta
# 需要utc时间
eta=datetime.utcnow() + timedelta(seconds=10)  # 延迟10秒
# eta=datetime.utcnow() + timedelta(days=1)  # 延迟1天
ret = tasks.low.apply_async(args=(200, 50), eta=eta)   # 通过args向任务函数low中传递参数
print(ret.id)    # 44bbf79c-0e34-41ed-a092-5d37e51d6cfa

get_result.py

from celery_task.celery import app

from celery.result import AsyncResult

id = '21325a40-9d32-44b5-a701-9a31cc3c74b5'
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('任务已经开始被执行')

高级使用:定时任务（一样使用包结构封装）

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

流程分析：

# 1）创建app + 任务

# 2）启动celery(app)服务：
# 非windows
# 命令：celery worker -A celery_task -l info
# windows：
# pip3 install eventlet
# celery worker -A celery_task -l info -P eventlet

# 3）添加任务：自动添加任务，所以要启动一个添加任务的服务，如果下面命令执行报错，就在末尾加-P eventlet
# 命令：celery beat -A celery_task -l info

# 4）获取结果

定时任务使用（需要在启动worker之外，再起一个beat，相当于雇佣的仆人来提交任我到broker）

celery.py

from celery import Celery

broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])


# 时区，使用亚洲上海时区，和下面UTC是否启用是一对，两者只能使用一个
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),    # 每三秒执行一次，beat就会向broker提交一次任务
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点
        'args': (300, 150),
    }
}

# 2 启动服务除了worker，还要启动beat
    # celery worker -A celery_task -l info -P eventlet
    # celery beat -A celery_task -l info

tasks.py

from .celery import app
import time
@app.task
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

get_result.py

from celery_task.celery import app

from celery.result import AsyncResult

id = '21325a40-9d32-44b5-a701-9a31cc3c74b5'
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,task,app,celery,任务,使用,print
来源： https://www.cnblogs.com/baicai37/p/13378954.html

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

ICode9

celery的使用

Celery

官方

Celery异步任务框架

Celery架构

消息中间件

任务执行单元

任务结果存储

使用场景

Celery的安装配置

两种celery任务结构：提倡用包管理，结构更清晰

Celery执行异步任务

基本结构

基本结构使用

celery_task.py

add_task.py

get_result.py

包架构封装（多任务结构）--推荐的使用方法

包架构封装使用

celery.py

tasks.py

add_task.py

get_result.py

高级使用:定时任务（一样使用包结构封装）

流程分析：

定时任务使用（需要在启动worker之外，再起一个beat，相当于雇佣的仆人来提交任我到broker）

celery.py

tasks.py

get_result.py