Python Celery分布式任务队列的安装与介绍(基于Redis)

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 --help
Options:
  -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.py


from 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.task
def 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, SUCCESS
result.get()         # 获取任务的返回值        
result.get(timeout=10)
result.get(propagate=False)  # 如果任务引发了异常, propagate=False表示异常不会被抛出来(默认情况会抛出来)
result.id            # 任务id 

注意事项:

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

Python Celery分布式任务队列的安装与介绍(基于Redis)
# -*- coding: utf-8 -*-
# @Time    : 2021/5/24 11:20
# @Author  : chinablue
# @File    : task.py


from 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.task
def add(x, y):
    return x + y
方式1: 实例化Celery时传入 Python Celery分布式任务队列的安装与介绍(基于Redis)
# -*- coding: utf-8 -*-
# @Time    : 2021/5/24 11:20
# @Author  : chinablue
# @File    : task.py


from 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.task
def add(x, y):
    return x + y
方式2: 通过conf的update方法 Python Celery分布式任务队列的安装与介绍(基于Redis)
# -*- coding: utf-8 -*-
# @Time    : 2021/5/24 11:20
# @Author  : chinablue
# @File    : task.py


from 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.task
def add(x, y):
    return x + y
方式3: 通过conf属性传递参数

 

上一篇:使用ProjectQ生成量子算法指令集


下一篇:Nginx配置go.conf