python-celery专注于实现分布式异步任务处理任务调度的插件！

Posted 2021-10-13 Python 集中营

celery是一个基于分布式消息传输的异步任务队列，它专注于实时处理，同时也支持任务调度。它的执行单元为任务（task），利用多线程，如Eventlet，gevent等，它们能被并发地执行在单个或多个职程服务器（worker servers）上。任务能异步执行（后台运行）或同步执行（等待任务成）。

【阅读全文】

Celery的三个主要模块

\'\'\'
worker：用于执行队列中的工作任务
beat：用于定时分配工作任务
task：工作任务，可以是在代码里的调用、也可以是被beat分配定时任务
\'\'\'

Celery的使用条件

\'\'\'
1、安装消息中间件的支持，一般是redis。
pip install redis
2、安装Celery的支持
pip install celery
3、安装本地redis服务端并启动
\'\'\'

做一个Celery定时任务

\'\'\'
导入相关依赖包
\'\'\'
import celery  # celery 定时任务包
import time  # 时间记录
from datetime import timedelta  # 时间处理

\'\'\'
创建一个worker用来执行定时任务
\'\'\'

# 创建worker、连接到redis消息中间件
worker = celery.Celery("work_test", backend="redis://localhost:6379/", broker="redis://localhost:6379/")

# 创建任务配置类TaskConfig

class TaskConfig:
    beat_suhedule = {
        \'update_info\': {
            \'task\': \'test013.test3.test_task\',  # 任务所在的模块.函数
            "schedule": timedelta(seconds=5),  # 定时任务间隔时间
        }
    }

# worker配置生效

worker.config_from_object(TaskConfig)


@worker.task
def test_task():
    \'\'\'
    任务
    :return:
    \'\'\'
    return "当前任务执行状态：success,{}".format(time.time())

\'\'\'
启动 beat 分配定时任务
celery -A test3.worker beat
\'\'\'

# C:\\work-code\\base-python3\\test013>celery -A test3.worker beat

\'\'\'
启动 beat 分配定时任务
celery -A test3.worker worker
\'\'\'

# C:\\work-code\\base-python3\\test013>celery -A test3.worker worker

【往期精彩】

● python远程服务操作工具：fabric，远程命令、本地命令、服务器操作利器！

● python超赞插件you-get，执行一行命令即可下载、命令行下载工具推荐！

● 办公自动化：Python-win32com自动将word文档转换成pdf格式！

● pandas数据统计插件的连接函数concat()妙用，灵活处理数据对象！

● Git LFS 3.0.0 发布，对大文件进行版本控制的 Git 扩展

● python有序序列的字典序列推导式运用技巧！

● Django 4.0 alpha 1 发布

● python经典有序序列的list列表推导式实践运用

● python常用转义字符串总结：各种字符转义的不同、如何取消转义字符效果？

● 如何使用python完成对WebService服务的调用？suds-py3插件安利一下！

● 介绍一款优秀的IDE Grid Studio，Excel深度集成python，直接编写并执行python代码块！

● python内置函数通过字符串的方式来执行函数代码块，类似java的反射机制相当强大！

● 磨刀不误砍柴工，PyCharm开发工具的常规配置，充分提高开发效率！

● python程序的处理进度、可视化管理，对运行步骤一目了然！

欢迎关注作者公众号【Python 集中营】，专注于后端编程，每天更新技术干货，不定时分享各类资料！

Celery—分布式的异步任务处理系统

Celery

1.什么是Clelery

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

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

同时也支持任务调度

Celery架构

 

Celery的架构由三部分组成：

● 消息中间件（message broker）

● 任务执行单元（worker）

● 任务执行结果存储（task result store）

消息中间件

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

任务执行单元

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

任务结果存储

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

 

 

 

 

版本支持情况

Celery version 4.0 runs on
Python ?2.7, 3.4, 3.5?
PyPy ?5.4, 5.5?
This is the last version to support Python 2.7, and from the next version (Celery 5.x) Python 3.5 or newer is required.

If you’re running an older version of Python, you need to be running an older version of Celery:

Python 2.6: Celery series 3.1 or earlier.
Python 2.5: Celery series 3.0 or earlier.
Python 2.4 was Celery series 2.2 or earlier.

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

 

2.使用场景

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

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

3.Celery的安装配置

pip install celery

消息中间件：RabbitMQ/Redis

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

 

4.Celery执行异步任务

基本使用

新建celery_task.py文件

# celery_task.py

from celery import Celery
# 不加密码
broker=‘redis://127.0.0.1:6379/0‘   # borker配置，任务队列
backend=‘redis://127.0.0.1:6379/1‘  # backend配置 执行结果存储

#加密码
# backend=‘redis://:[email protected]:6379/1‘
# broker=‘redis://:[email protected]:6379/2‘
#一定要指定一个名字


app=Celery(‘test‘,broker=broker,backend=backend)  # 创建任务异步处理器


#任务其实就是个函数
#需要用一个装饰器 *.task装饰,表示该任务是被 * celery管理的，并且可以用celery执行的
@app.task
def add(x,y):
    import time
    time.sleep(2)  # 模拟阻塞任务
    return x+y

这样我们就完成了上图流程中的三大配置，下面我们需要去提交任务并启动worker执行任务

另建add_task.py文件

#用于提交任务的py文件

import celery_task

#提交任务到消息队列中
#只是把任务提交到消息队列中，并没有执行需要启动worker才可以生效
ret=celery_task.add.delay(3,4)  # add 即celery_task中add函数
print(ret)
# ret=a5ea035f-0cc3-44ba-b334-f5d7c7ce681d  :任务的id号，待任务执行完毕，需要通过这个id去backend去取执行结果



#提交定时任务：于2019-07-20 11:13:56执行的任务，此任务只是在这里提交了，需要启动worker才可以生效
from datetime import datetime
v1 = datetime(2019, 7, 20, 11, 13, 56)
print(v1)  # 时间对象
v2 = datetime.utcfromtimestamp(v1.timestamp())
print(v2)  

# #取出要执行任务的时间对象，调用apply_async方法，args是参数，eta是执行的时间
result = celery_task.add.apply_async(args=[1, 3], eta=v2)
print(result.id)   # 任务的id号，待任务执行完毕，需要通过这个id去backend去取执行结果

#第二种获取时间的方法
ctime = datetime.now()
# 默认用utc时间
utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
from datetime import timedelta
#取10s之后的时间对象
time_delay = timedelta(seconds=3)
task_time = utc_ctime + time_delay   # 时间对象的相加
result = celery_task.add.apply_async(args=[4, 3], eta=task_time)
print(result.id)

启动worker

启动worker的两种方法：

# 方法一：
from celery_task import app
if __name__ == ‘__main__‘:
    app.worker_main()
    # cel.worker_main(argv=[‘--loglevel=info‘)

# 方法二：
# 命令行启动—常用
# linux下： celery worker -A celery_task_s1 -l info
# windows下：celery worker -A celery_task_s1 -l info -P eventlet  #eventlet是个模块 需要pip装下

 

接下来就是查看任务执行结果了

另起result.py文件

# 创建py文件：result.py，查看任务执行结果

from celery.result import AsyncResult
from celery_task import app

async = AsyncResult(id="e919d97d-2938-4d0f-9265-fd8237dc2aa3", app=cel)

if async.successful():
    result = async.get()
    print(result)
    # result.forget() # 将结果删除
elif async.failed():
    print(‘执行失败‘)
elif async.status == ‘PENDING‘:
    print(‘任务等待中被执行‘)
elif async.status == ‘RETRY‘:
    print(‘任务异常后正在重试‘)
elif async.status == ‘STARTED‘:
    print(‘任务已经开始被执行‘)

 

总结下Celery使用步骤：

创建celery_app_task.py，配置borker\\backend\\worker和任务函数

执行 add_task.py，添加任务，并获取任务ID

执行命令启动worker：celery worker -A celery_app_task -l info

执行 result.py,检查任务状态并获取结果

 

 下面我们进入重点：