怎么能不恨呢，在我发现自己是恶鬼的时候，在我最绝望最虚弱的时候，这个世上最该跟我在一起的人却用刀把我的心刺穿了

Celery 是 Distributed Task Queue，分布式任务队列。分布式决定了可以有多个 worker 的存在，队列表示其是异步操作。它可以让任务的执行完全脱离主程序，甚至可以被分配到其他主机上运行。我们通常使用它来实现异步任务（async task）和定时任务（crontab）。

安装

Celery4.x 开始不再支持Windows平台了。下面装的是3.1.25。

安装使用命令

pip3 install celery==3.1.25

查看是否安装成功，使用命令即可：

celery --version

如果在win10下想要使用celery4.x的话，可以这么做：

pip3 install eventlet

运行worker的时候加上一个参数：

celery -A xxxx worker -l info -P eventlet

然后安装redis(个人比较喜欢redis)

首先下载安装redis windowns服务端

下载地址

解压后，在其目录下执行命令

redis-server.exe

即可启动redis数据库

然后安装python连接操作redis的库

pip3 install redis==2.10.6

注意版本号

核心主件

celery通过五大模块实现

Task

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

Broker

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

Worker

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

Beat

定时任务调度器，根据配置定时将任务发送给Broler。

Backend

用于存储任务的执行结果。

组成关系

各个角色间的关系看下面这张图理解一下：

初次使用

首先编写一个文件命名为task1.py

from celery import Celery
app = Celery('tasks',broker='redis://192.168.1.102:6379/0')
# redis://192.168.1.102:6379/0 是redis数据库地址，无需账号密码验证，也是ssrf在获取内网系统权限的方式之一
@app.task
def add(x,y):
    print('传递 {} + {} = {}'.format(x,y,x+y))
    return x+y

然后启动redis数据库

接下来再task1文件夹执行命令

celery -A task1 worker --loglevel=info

就会看到消息队列都启动

到现在所有的队列都启动，可以向这个队列添加任务等待处理

方法是再task1目录下打开cmd窗口，进入python3交互界面

python3
from task1 import add
add.delay(6,12)
add.delay(6,6)

上面只是一个发送任务的调用，结果是拿不到的。上面也没有接收返回值，这次把返回值保存到起来

修改task1内容

app = Celery('tasks',broker='redis://192.168.1.102:6379/0',backend='redis://192.168.1.102:6379/0')

然后要重启Worker，IDLE也要重启

然后这样就能获取结果了

t = add.delay(1, 1)
t.get()
# 还可以设置超时时间 t.get(timeout=5)

# 如果出错，获取错误结果，不触发异常
# 使用命令t.get(propagate=False)    
# t.traceback  （打印异常详细结果）

# 还可以获取任务状态
# t.ready() 返回True 或者False

在项目中使用Celery

可以把celery配置成一个应用，假设应用名字是CeleryPro，目录格式如下：

CeleryPro
├─__init.py
├─celery.py
├─tasks.py

这里的连接文件命名必须为celery.py，其他名字随意

celery文件

这个文件名必须是celery.py：

from __future__ import absolute_import, unicode_literals
from celery import Celery

app = Celery('CeleryPro',
             broker='redis://192.168.1.102:6379',
             backend='redis://192.168.1.102:6379',
             include=['CeleryPro.tasks'])
app.conf.update(
    result_expires=3600,
)

if __name__ == '__main__':
    app.start()

tasks文件

这个文件开始两行就多了一个点，这里要导入上面的celery.py文件。后面只要写各种任务加上装饰器就可以了：

from __future__ import absolute_import, unicode_literals
from .celery import app
import time

@app.task
def add(x, y):
    print("计算2个值的和: %s %s" % (x, y))
    return x+y

@app.task
def upper(v):
    for i in range(10):
        time.sleep(1)
        print(i)
    return v.upper()

启动worker

启动的时候，-A 参数后面用应用名称 CeleryPro 。你还需要cd到你CeleryPro的父级目录上启动，否则找不到：

启动的姿势

这里注意用的都是CeleryPro：

celery -A CeleryPro worker -loglevel=info  # 前台启动不推荐
celery -A CeleryPro worker -l info  # 前台启动简写
celery multi start w1 -A  CeleryPro -l info  # 推荐用后台启动

定时任务

主要修改 celery.py文件

from __future__ import absolute_import, unicode_literals
from celery import Celery
from celery.schedules import crontab
from datetime import timedelta

app = Celery('CeleryPro',
             broker='redis://192.168.1.102',
             backend='redis://192.168.1.102',
             include=['CeleryPro.tasks'])

app.conf.CELERYBEAT_SCHEDULE = {
    'add every 10 seconds': {
        'task': 'CeleryPro.tasks.add',
        'schedule': timedelta(seconds=10),  
        # 可以用timedelta对象
        # 'schedule': 10,  # 也支持直接用数字表示秒数
        'args': (1, 2)
    },
    'upper every 2 minutes': {
        'task': 'CeleryPro.tasks.upper',
        'schedule': crontab(minute='*/2'),
        'args': ('abc', ),
    },
}

# app.conf.CELERY_TIMEZONE = 'UTC'
app.conf.CELERY_TIMEZONE = 'Asia/Shanghai'

# Optional configuration, see the application user guide.
app.conf.update(
    CELERY_TASK_RESULT_EXPIRES=3600,
)

if __name__ == '__main__':
    app.start()

启动使用命令

celery -A CeleryPro beat -l info
celery -A CeleryPro worker -l info

参数解析：

-l info     与--loglevel=info的作用是一样的。
--beat    周期性的运行。即设置 心跳。

新例子

# tasks.py
# coding:utf-8
from celery import Celery,platforms

app = Celery('tasks')
app.config_from_object('config')
platforms.C_FORCE_ROOT = True

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

和另一个文件

# config.py
# coding:utf-8
from __future__ import absolute_import
from celery.schedules import crontab
from datetime import timedelta

BROKER_URL = 'redis://127.0.0.1:6379/0'

CELERYBEAT_SCHEDULE = {
    'add-every-2-seconds': {
        'task': 'tasks.add',
        'schedule': timedelta(seconds=2),
        'args': (16, 10),
    },
}

CELERY_TIMEZONE = 'Asia/Shanghai'

然后打开三个cmd窗口，依次输入：

celery -A tasks beat -l info
celery -A tasks worker -l info
celery -A tasks flower

然后访问本地5555端口即可~

查看异步任务情况

Celery提供了一个工具flower，将各个任务的执行情况、各个worker的健康状态进行监控并以可视化的方式展现

下实现的方式如下：

安装flower:

pip3 install flower

启动flower（默认会启动一个webserver，端口为5555）:

在另一个Terminal中：

celery -A task1 flower

这里的task1是上面创建的py文件

进入

http://localhost:5555

即可查看。

资料文档

致力于Python开发网络安全工具,分享Python底层与进阶知识，漏洞扫描器开发与爬虫开发

分布式消息队列-Celery

安装