Python Celery分布式任务队列的安装方法

本篇内容介绍了“Python Celery分布式任务队列的安装方法”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！

Celery是一个基于Python编写的分布式任务队列(Distributed Task Queue), 通过对Celery进行简单操作就可以实现任务(耗时任务, 定时任务)的异步处理

一. Celery的安装

Celery4.0版本开始,不支持windows平台

1.1 通过pip方式安装celery

pip install -U "Celery[redis]"

注意事项:

　　在windows上安装后,可能会出现如下报错:

ValueError: '__name__' in __slots__ conflicts with class variable

 　　此时先卸载celery, 然后尝试通过如下命令重新进行安装

pip install -U https://github.com/celery/py-amqp/zipball/master
pip install -U https://github.com/celery/billiard/zipball/master
pip install -U https://github.com/celery/kombu/zipball/master
pip install -U https://github.com/celery/celery/zipball/master
pip install -U "Celery[redis]"

1.2 给celery创建一个软连接

ln -s ~/.venv/project_dj/bin/celery /usr/bin/celery

1.3 执行celery命令

[root@localhost ~]$ celery --helpOptions:  -A, --app APPLICATION  -b, --broker TEXT  --result-backend TEXT  --loader TEXT  --config TEXT  --workdir PATH  -C, --no-color  -q, --quiet  --version  --help                 Show this message and exit.Commands:
  amqp     AMQP Administration Shell.
  beat     Start the beat periodic task scheduler.
  call     Call a task by name.
  control  Workers remote control.
  events   Event-stream utilities.
  graph    The ``celery graph`` command.
  inspect  Inspect the worker at runtime.
  list     Get info from broker.
  logtool  The ``celery logtool`` command.
  migrate  Migrate tasks from one broker to another.
  multi    Start multiple worker instances.
  purge    Erase all messages from all known task queues.
  report   Shows information useful to include in bug-reports.
  result   Print the return value for a given task id.
  shell    Start shell session with convenient access to celery symbols.
  status   Show list of workers that are online.
  upgrade  Perform upgrade between versions.  worker   Start worker instance.

二. Celery的基本使用

2.1 创建celery应用, 并定义任务

# -*- coding: utf-8 -*-# @Time    : 2021/5/24 11:20# @Author  : chinablue# @File    : task.pyfrom celery import Celery# 创建一个app(Celery实例),作为所有celery操作的切入点broker_url = f"redis://:123456@127.0.0.1:6379/5"backend_url = f"redis://:123456@127.0.0.1:6379/6"app = Celery("tasks", broker=broker_url, backend=backend_url)# 定义一个任务@app.taskdef add(x, y):return x + y

事项说明：

　　1) 创建Celery实例时，需要指定一个消息代理(broker)来接收和发送任务消息. 本文使用的是Redis(docker redis搭建)

　　2) broker和backend参数的格式: redis://:password@hostname:port/db_number

2.2 启动celery worker服务端

celery -A tasks worker --loglevel=INFO

事项说明：

　　1) 在生产环境中, 会使用supervisor工具将celery服务作为守护进程在后台运行

2.3 调用任务

打开终端, 进入python命令行模式:

>>> result = add.delay(4, 4)
>>> result = add.apply_async((4, 4), countdown=5)

事项说明：

　　1) add.apply_async((4, 4)) 可以简写为 add.delay(4, 4)

　　2) add.apply_async((4, 4), countdown=5) 表示任务发出5秒后再执行

2.4 追踪任务信息

若想获取每个任务的执行信息,在创建Celery实例时, 需要指定一个后端(backend). 本文使用的是Redis(docker redis搭建)

result = add.delay(4, 4)        
result.ready()       # 任务状态: 进行中, 已完成result.failed()      # 任务完成, 任务失败result.successful()  # 任务完成, 任务成功result.state         # 任务状态: PENDING, STARTED, SUCCESSresult.get()         # 获取任务的返回值        result.get(timeout=10)
result.get(propagate=False)  # 如果任务引发了异常, propagate=False表示异常不会被抛出来(默认情况会抛出来)result.id            # 任务id

注意事项：

　　1) 在celery中，如果想配置backend参数，有如下三种方式

# -*- coding: utf-8 -*-# @Time    : 2021/5/24 11:20# @Author  : chinablue# @File    : task.pyfrom celery import Celery# 创建一个app(Celery实例),作为所有celery操作的切入点broker_url = f"redis://:123456@127.0.0.1:6379/5"backend_url = f"redis://:123456@127.0.0.1:6379/6"app = Celery("tasks", broker=broker_url, backend=backend_url)# 定义一个任务@app.taskdef add(x, y):return x + y

方式1: 实例化Celery时传入

# -*- coding: utf-8 -*-# @Time    : 2021/5/24 11:20# @Author  : chinablue# @File    : task.pyfrom celery import Celery

broker_url = f"redis://:123456@127.0.0.1:6379/5"backend_url = f"redis://:123456@127.0.0.1:6379/6"app = Celery("tasks")

app.conf.update({"broker_url": broker_url,"result_backend": backend_url,
})# 定义一个任务@app.taskdef add(x, y):return x + y

方式2: 通过conf的update方法

# -*- coding: utf-8 -*-# @Time    : 2021/5/24 11:20# @Author  : chinablue# @File    : task.pyfrom celery import Celery

broker_url = f"redis://:123456@127.0.0.1:6379/5"backend_url = f"redis://:123456@127.0.0.1:6379/6"app = Celery("tasks")

app.conf.broker_url = broker_url
app.conf.result_backend = backend_url# 定义一个任务@app.taskdef add(x, y):return x + y

“Python Celery分布式任务队列的安装方法”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注亿速云网站，小编将为大家输出更多高质量的实用文章！