python中celery的基本使用详情

 

1.基本介绍

Celery 是由Python 编写的简单,灵活,可靠的用来处理大量信息的分布式系统,它同时提供操作和维护分布式系统所需的工具。Celery 专注于实时任务处理,支持任务调度。

简单的说,它就是一个分布式队列的管理工具,用celery提供的接口快速实现并管理一个分布式的任务队列。

有一点我们需要搞清楚,Celery 本身并不是任务队列,它是一个分布式队列的管理工具,Celery封装好了操作常见任务队列的各种操作,比如说从监听某个任务队列并从该队列中拿到数据进行消费。

 

2.使用场景

它可以让任务的执行完全脱离主程序,甚至可以被分配到其他主机上运行。我们通常使用它来实现异步任务(async task)和定时任务(crontab)。

  • 异步任务: 将耗时操作任务提交给Celery去异步执行,比如发送短信/邮件、消息推送、音视频处理等等
  • 定时任务: 定时执行某件事情,比如每天数据统计

 

3.工作流程和组成部分

这里用一张图片说明下:

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

消息中间件:

Celery本身不提供消息服务,但是可以方便的和第三方提供的消息中间件集成。包括RabbitMQ, Redis等等,官方推荐用rabbitMQ,因为它持久稳定。

任务执行单元:

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 等等 先安装模块

pip install celery
pip install redis

 

4.Celery执行异步任务

4.1 基础使用

这里项目结构如下:

第一步:先创建celery相关配置配置celery_object.py

import celery

# 执行如下命令: celery -A celery_object worker -l info

backend = "redis://127.0.0.1:6379/4"  # 设置redis的4号数据库来存放结果
broker = "redis://127.0.0.1:6379/5"  # 设置redis的5号数据库存放消息中间件
celery_app = celery.Celery(
  "celery_demo",
  backend=backend,
  broker=broker,
  include=[
      "celery_task",
  ],
)

celery_app.conf.task_serializer = "json"
celery_app.conf.result_serializer = "json"
celery_app.conf.accept_content = ["json"]

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

参数说明:

  • backend 就是异步任务执行完成以后,结果存放的地方。
  • broker 就是具体执行任务的工作节点。
  • celery.Celery()方法是实例化一个celery对象。

第二步:创建任务相关的文件celery_task.py

import time

from celery_object import celery_app

@celery_app.task
def send_email(name):
  print("向%s发送邮件..." % name)
  time.sleep(5)
  print("向%s发送邮件完成" % name)
  return f"成功拿到{name}发送的邮件!"

@celery_app.task
def send_msg(name):
  print("向%s发送短信..." % name)
  time.sleep(5)
  print("向%s发送短信完成" % name)
  return f"成功拿到{name}发送的短信!"

通过@celery_app.task这样的装饰器,成功的把对应的函数变成对应celery的异步worker函数。

紧接着我们在项目当前所在的目录执行命令:

celery -A celery_object worker -l info
  • -A 指的是application应用对象
  • worker 就是工作人(固定写法)
  • -l 指的是日志级别,这里是打印info级别的日志

之后就可以有下面的输出显示就代表celery启动成功:

之后我们就可以向celery生产任务了,创建produce_result.py文件。

from celery_task import send_email, send_msg

if __name__ == "__main__":
  for i in range(10):
      result = send_email.delay(f"张三{i}")
      print(result.id)
      result2 = send_msg.delay(f"李四{i}")
      print(result2.id)

运行生产任务的程序,会看到如下的数据,这里打印的就是任务ID。

然后在终端可以看到下面的东西,就代表celery成功的拿到队列中任务 并进行消费了。

然后打开我们的redis可以看到有对应的数据记录。

与此同时 我们还可以查看celery任务ID的状态,check_result.py写入如下:

from celery.result import AsyncResult
from celery_object import celery_app

async_result = AsyncResult(id="d1c722fa-4ebf-432e-967e-a462bdefeac4", app=celery_app)
print("任务状态:", async_result.status)
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("任务已经开始被执行")

运行结果:

任务状态: SUCCESS
成功拿到李四0发送的短信!

关于python中celery的基本使用详情的文章就介绍至此,更多相关python celery内容请搜索编程宝库以前的文章,希望以后支持编程宝库

 一、re模块简介Python 的 re 模块(Regular Expression 正则表达式)提供各种正则表达式的匹配操作,和 Perl 脚本的正则表达式功能类似,使用这一内嵌 ...