任务队列 Celery

Posted 2022-11-21 PolluxAvenger

我们需要一个任务队列来进行任务的分配和维护，我们选择最常见的 Celery

Celery 介绍

Celery (芹菜)是一个异步任务队列/基于分布式消息传递的作业队列
（PS：不得不说，我并不喜欢芹菜，这差点让我放弃了它

Celery 的结构非常简单，大致分为 3 个部分：

• worker部分负责任务的处理，即执行工作的线程
• broker部分负责任务消息的分发以及任务结果的存储，这部分任务主要由中间数据存储系统完成，比如消息队列服务器 RabbitMQ 、Redis 、Amazon SQS、MongoDB、IronMQ 等或者关系型数据库，使用关系型数据库依赖 Sqlalchemy 或者 Django 的 ORM
• Celery主类负责进行任务最开始的指派与执行控制，可以是单独的 Python 脚本，也可以和其他程序结合，应用到 Django 或者 Flask 等 Web 框架里面以及你能想到的任何应用

安装

pip install celery
pip install celery-with-redis
pip install flower

flower 是 Celery 的监控工具,提供了图形界面监控任务的执行

安装 redis

wget http://download.redis.io/releases/redis-4.0.9.tar.gz
tar xzf redis-4.0.9.tar.gz
cd redis-4.0.9
make -j8

启动 redis

src/redis-server

启动 flower

python2.7 /usr/local/bin/flower --port=7788 --broker=redis://localhost:6379/0 --broker_api=redis://localhost:6379/0

启动 celery

celery -A tasks worker --loglevel=info

提交任务

tasks.py

import time
from celery import Celery, platforms

platforms.C_FORCE_ROOT = True
celery = Celery("tasks", broker='redis://localhost:6379/0')

@celery.task
def add(x, y):
    return x + y

任务执行

from tasks import *

add.delay(1, 2)

执行结果

查看 flower(http://localhost:7788/)

以 root 身份启动 celery

celery -A symbolic_exec worker --loglevel=info
export C_FORCE_ROOT = "true"

若任务被修改，Celery 需要重新启动，否则无法随之更改