Celery 初探

Celery是一个分布式任务队列。它是围绕最佳实践设计的，这样您的产品就可以与其他语言进行伸缩和集成，而且它还提供了在生产环境中运行这样一个系统所需的工具和支持。

Celert将包括几部分内容：

• 选择和安装消息传输（代理）
• 安装Celery并创建第一个任务
• 启动工作和调用工作
• 跟踪在不同状态下的任务并检查返回值

Celery设计是故意保持简单，这样就不会和高级特性混淆。

选择一个代理（中间件）

Celery需要一个发送和接收消息的解决方案，通常是以一个单独的服务出现的，称为消息代理（消息中间件）

RabbitMQ

RabbitMQ功能齐全、稳定、耐用并且容易安装。对于生产环境来说是一个很好的选择。

安装命令：

sudo yum install rabbitmq-server // Centos
sudo apt-get install rabbitmq-server // Ubuntu or Debian

Redis

Redis也功能齐全，但是在突然中止或者电源故障的情况下更容易出现数据丢失

安装Celery

Celert在Python包索引上，可以直接通过pip或者easy_install安装：

pip install celery

应用

第一件事情需要一个Celery实例。我们称之为Celery Application或者简称app

由于这个实例是你想在Celery中想做的所有事情的入口点，比如创建任务和管理人员，必须在其他模块中导入它。

创建一个文件 tasks.py:

from celery import Celery

app = Celery('tasks', broker='pyamqp://guest@localhost//')

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

对Celery而言，第一个参数是当前模块的名称，只有这样，当任务在__main__模块中定义时，才能自动生成名称。

第二个参数是代理关键字参数，指定你想用的消息代理的URL。上述示例中用的是Redis

对于RabbitMQ 使用amqp://localhost，对于Redis使用redis://localhost

至此，定义了一个叫做add的简单任务，返回两个数的和。

运行Celery工作服务器

使用worker参数来执行程序

celery -A tasks worker --loglevel=info

生产环境中，将在后台运行worker作为守护进程

或者可用命令行的完整清单：

celery worker --help

获取其他可用命令

celery help

调用方法

使用delay()方法调用任务

这是对apply_async()方法的一个便捷的快捷方式，可以更好地控制任务的执行：

from tasks import add
add.delay(4, 4)

任务交给之前worker处理，可以在worker的输出控制台查看

调用一个任务返回一个AsyncResult实例，它可以被用来检查任务的状态、等待任务完成或者得到它的返回值（如果任务失败获得异常信息）

结果在默认中是不可用的。为了进行远程程序调用或者跟踪数据库中的任务结果，可以配置Celery来使用result backend。

保存结果

如果想跟踪任务状态，Celery需要存储或者发送状态到某个地方。有几种内置result backend选择：

• SQLAlchemy/Django ORM
• Memcached
• Redis
• RPC(RabbitMQ/AMQP)

下例中间使用rpc作为result backend，它将状态作为临时消息返回给后端。后端由backend参数配置Celery，或者由result_end设置。

app = Celery('tasks', backend='rpc://', broler='pyamqp://')

或者使用Redis作为result backend，但人使用RabbitMQ作为消息代理（一种流行的结合）：

app = Celery('tasks', backend='redis://loalhost', broker='pyamqp;//')

result backend配置好了，再执行一次任务

result = add.delay(4, 4)

• ready() 返回任务是否完成（boolean）

result.ready()

• get() 等待结果完成返回结果（不推荐：将异步调用编程了同步）

result.get(timeout=1)

如果任务引发了异常，get()将重新引发异常，可以通过声明propagete参数覆盖

result.get(propagate=False)

propagate eg:传播

如果任务抛出异常，可以访问原始的错误消息：

result.traceback

没有错误的情况下返回

AttributeError: 'AsyncResult' object has no attribute 'trackback'

配置

• Celey，不需要太多配置来操作。它只有一个输入和输出，输入必须连接一个消息中间件（消息代理），输出可以选择是否连接一个result backend

• 对于大多数用例，默认配置就已经很好了，当然也可以根据自己的情况按需选择性配置。了解配置选项可以帮助了解熟悉Celery。

• 配置可以直接在app中设置，也可以在单独的模块中设置（例如：通过task_serializer来配置序列化任务负载的缺省序列化器 app.conf.task_serializer = 'json'）

• 如果有许多配置可以使用update:

app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='Asia/Shanghai' # UTF+8
    enable_utc=True,
)

• 对于大项目，推荐使用专门的配置模块。不提倡硬编码周期任务间隔和任务路由选项，把这些放在一个集中的位置会更好，对于Libraries来说更是如此，它能使用户控制他们任务的行为。集中式配置也允许系统管理员在系统出现问题的时候做简单的更改。

• 调用app.conf_from_object()方法可以告诉Celery实例用配置模块

app.config_from_object('celeryconfig')

这个模块通常叫做celeryconfig，当然也可以用任何模块名命名。

• 根据以上情况，一个名为celeryconfig.py的模块必须能从当前路径或者Python的路径加载。例如（celeryconfig.py）：

broker_url = 'pyamqp://'
result_backend = 'rpc://'

task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json']
timezone = 'Asia/Shanghai'
enable_utc = True

为了验证配置文件是否正确并且生效，可以使用

python -m celeryconfig

为了演示配置文件的强大，这是如何将一个错误的任务队列路由到一个专用的队列上：celeryconfig.py

task_routes = {
    'task.add': 'low-priority'
}

或者可以对任务进行限制，而不是对它进行路由，这样就可以在一分钟诶处理这个类的10个任务（10/m)celeryconfig.py:

task_annotions = {
    'tasks.add': {'rate_limit': '10/m'}
}

如果使用RabbitMQ或者Redis来做为中间件，可以通过指导worker在运行时为任务设置一个速度限制
celery -A tasks control rate_limit tasks.add 10/m