celery-分布式任务调度

Posted 2023-02-19 程序员唐丁

唐丁最近工作比较忙，好久没更新了。最近一直在做Python的Web后端开发工作。所以这次就介绍一款后端开发常用的一个分布式任务调度框架–Celery。
介绍

Celery是Python开发的分布式任务调度模块，Celery本身不含消息服务，它使用第三方消息服务来传递任务，目前，Celery支持的消息服务有RabbitMQ、Redis甚至是数据库。
安装

pip install Celery
核心模块

Task : 就是任务，有异步任务和定时任务

Broker ： 中间人，接收生产者发来的消息即Task，将任务存入队列。任务的消费者是Worker，Celery本身不提供队列服务，推荐用Redis或RabbitMQ实现队列服务

Worker : 执行任务的单元，它实时监控消息队列，如果有任务就获取任务并执行它

Beat : 定时任务调度器，根据配置定时将任务发送给Broker

Backend : 用于存储任务的执行结果

常用命令

配置celery – celery.py

from celery import Celery
app = Celery(‘proj’,
broker=‘amqp://’,
backend=‘rpc://’,
include=[‘proj.tasks’])

Optional configuration, see the application user guide.

app.conf.update(
result_expires=3600,
)

从相关模块中自动加载任务

app.autodiscover_tasks(settings.INSTALLED_APPS)
​
#定义任务 – tasks.py
from .celery import app
@app.task
def add(x, y):
return x + y
@app.task
def mul(x, y):
return x * y
@app.task
def xsum(numbers):
return sum(numbers)
​

后台一个或多个启动

celery multi start w1 -A proj -l INFO

后台重启

celery multi restart w1 -A proj -l INFO

后台立即停止

celery multi stop w1 -A proj -l INFO

后台等待所有程序运行结束停止

celery multi stopwait w1 -A proj -l INFO
​

指定发送任务到指定队列-在配置文件中

task_routes=(
‘proj.tasks.my_task5’: ‘queue’: ‘queue1’,
‘proj.tasks.my_task6’: ‘queue’: ‘queue1’,
‘proj.tasks.my_task7’: ‘queue’: ‘queue2’,
,
)
​

调用任务

delay()
apply_async((2, 2), queue=‘lopri’, countdown=10)
在上面的例子中，任务将被发送到一个名为的队列lopri，任务将在消息发送后最早 10 秒执行。

celery配置

定义任务-tasks.py

import time
from celery import Celery
​
celery = Celery(‘tasks’, broker=‘redis://localhost:6379/0’)
​
@celery.task
def sendmail(mail):
print(‘sending mail to %s…’ % mail[‘to’])
time.sleep(2.0)
print(‘mail sent.’)
​

启动worker

celery -A tasks worker --loglevel=info

发送任务到broker

from tasks import sendmail
sendmail.delay(dict(to=‘[e-mail]’))