Celery 简介

Introduction to Celery

什么是任务队列？

What's a Task Queue?

中文

任务队列用于在不同的线程或机器之间分发工作负载。

任务队列的输入是一种称为任务（task）的工作单元。专门的 worker 进程会持续监控任务队列，以获取新的工作任务来执行。

Celery 通过消息进行通信，通常使用一个消息代理（broker）在客户端与 worker 之间进行协调。要启动一个任务，客户端会向队列中添加一条消息，然后 broker 会将该消息传递给某个 worker。

一个 Celery 系统可以由多个 worker 和 broker 组成，从而实现高可用性（HA）和水平扩展（horizontal scaling）。

Celery 使用 Python 编写，但其协议可以由任何语言实现。除了 Python，还有适用于 Node.js 的 node-celery 和 node-celery-ts，以及一个 PHP 客户端。

也可以通过暴露 HTTP 接口，并让任务发起请求（即 webhook）来实现不同语言之间的互操作。

英文

Task queues are used as a mechanism to distribute work across threads or machines.

A task queue's input is a unit of work called a task. Dedicated worker processes constantly monitor task queues for new work to perform.

Celery communicates via messages, usually using a broker to mediate between clients and workers. To initiate a task the client adds a message to the queue, the broker then delivers that message to a worker.

A Celery system can consist of multiple workers and brokers, giving way to high availability and horizontal scaling.

Celery is written in Python, but the protocol can be implemented in any language. In addition to Python there's node-celery and node-celery-ts for Node.js, and a PHP client.

Language interoperability can also be achieved exposing an HTTP endpoint and having a task that requests it (webhooks).

我需要什么？

What do I need?

中文

Celery 需要一个消息传输机制来发送和接收消息。 RabbitMQ 和 Redis 的 broker 传输实现是功能完整的， 此外还有许多其他实验性解决方案可用，包括使用 SQLite 进行本地开发。

Celery 可以运行在单台机器上、多台机器上，甚至可以跨数据中心运行。

英文

Celery requires a message transport to send and receive messages. The RabbitMQ and Redis broker transports are feature complete, but there's also support for a myriad of other experimental solutions, including using SQLite for local development.

Celery can run on a single machine, on multiple machines, or even across data centers.

开始使用

Get Started

中文

如果这是你第一次使用 Celery，或者你在 3.1 版本之后没有持续关注其发展（即从更早版本迁移过来）， 那么你应该首先阅读我们的入门教程：

• 使用 Celery 的第一步

• 下一步

英文

If this is the first time you're trying to use Celery, or if you haven't kept up with development in the 3.1 version and are coming from previous versions, then you should read our getting started tutorials:

• 使用 Celery 的第一步

• 下一步

Celery 是……

Celery is…

中文

• 简单易用

  Celery 易于使用和维护，且 无需配置文件。

  Celery 拥有一个活跃而友好的社区， 你可以通过 `邮件列表`_ 和 IRC 频道 寻求支持。

  下面是一个最简单的应用示例：

  from celery import Celery
  
  app = Celery('hello', broker='amqp://guest@localhost//')
  
  @app.task
  def hello():
      return 'hello world'
  

• 高可用性

  在连接丢失或失败时，worker 和客户端会自动重试连接， 而且某些 broker 支持 主主（Primary/Primary） 或 主备（Primary/Replica） 复制机制。

• 高性能

  单个 Celery 进程可以每分钟处理数百万个任务， 往返延迟（round-trip latency）可低于毫秒级（使用 RabbitMQ、librabbitmq 以及经过优化的配置）。

• 高度灵活

  Celery 几乎所有部分都可以被扩展或单独使用， 包括自定义的进程池实现、序列化器、压缩方案、日志记录器、 调度器（schedulers）、消费者（consumers）、生产者（producers）、broker 传输机制等等。

英文

• Simple

  Celery is easy to use and maintain, and it doesn't need configuration files.

  It has an active, friendly community you can talk to for support, including a mailing-list and an IRC channel.

  Here's one of the simplest applications you can make:

  from celery import Celery
  
  app = Celery('hello', broker='amqp://guest@localhost//')
  
  @app.task
  def hello():
      return 'hello world'
  

• Highly Available

  Workers and clients will automatically retry in the event of connection loss or failure, and some brokers support HA in way of Primary/Primary or Primary/Replica replication.

• Fast

  A single Celery process can process millions of tasks a minute, with sub-millisecond round-trip latency (using RabbitMQ, librabbitmq, and optimized settings).

• Flexible

  Almost every part of Celery can be extended or used on its own, Custom pool implementations, serializers, compression schemes, logging, schedulers, consumers, producers, broker transports, and much more.

支持以下功能 / It supports

中文

消息代理（Brokers） RabbitMQ、Redis， Amazon SQS，以及更多…… 并发模式（Concurrency） prefork（多进程） Eventlet、gevent thread（多线程） `solo`（单线程）

结果存储（Result Stores） AMQP、Redis Memcached SQLAlchemy、Django ORM Apache Cassandra、Elasticsearch、Riak MongoDB、CouchDB、Couchbase、ArangoDB Amazon DynamoDB、Amazon S3 Microsoft Azure Block Blob、Microsoft Azure Cosmos DB Google Cloud Storage 文件系统（File system） 序列化（Serialization） pickle、json、yaml、msgpack zlib、bzip2 压缩 消息加密签名（Cryptographic message signing）

英文

Brokers RabbitMQ, Redis, Amazon SQS, and more… Concurrency prefork (multiprocessing), Eventlet, gevent thread (multithreaded) solo (single threaded)

Result Stores AMQP, Redis Memcached, SQLAlchemy, Django ORM Apache Cassandra, Elasticsearch, Riak MongoDB, CouchDB, Couchbase, ArangoDB Amazon DynamoDB, Amazon S3 Microsoft Azure Block Blob, Microsoft Azure Cosmos DB Google Cloud Storage File system Serialization pickle, json, yaml, msgpack. zlib, bzip2 compression. Cryptographic message signing.

功能

Features

中文

监控（Monitoring） worker 会发出一系列监控事件， 内置工具和外部工具可以利用这些事件实时了解集群的运行状况。 阅读更多…。 工作流（Work-flows） 可以使用一套我们称之为 "canvas" 的强大原语（primitives）， 来组合简单或复杂的工作流， 包括分组（grouping）、链式调用（chaining）、分块处理（chunking）等。 阅读更多…。 时间与速率限制（Time & Rate Limits） 你可以控制每秒/每分钟/每小时执行的任务数量， 也可以限制单个任务的最长运行时间， 这些限制可以设置为默认值、应用于特定 worker， 或单独应用于每种任务类型。 阅读更多…。

调度（Scheduling） 你可以通过秒数或 datetime 指定任务的运行时间， 也可以使用周期性任务（periodic tasks）， 基于简单的时间间隔或 Crontab 表达式进行重复调度， 支持按分钟、小时、星期几、每月几号和每年几月进行设置。 阅读更多…。 资源泄漏保护（Resource Leak Protection） 当用户任务发生资源泄漏（如内存或文件描述符泄漏）且无法控制时， 可以使用 --max-tasks-per-child 选项来进行保护。 阅读更多…。 用户自定义组件（User Components） 每个 worker 组件都可以自定义， 用户还可以定义额外的组件。 worker 是通过一组称为 "bootsteps" 的依赖关系图构建起来的， 允许对 worker 内部行为进行细粒度控制。

英文

Monitoring A stream of monitoring events is emitted by workers and is used by built-in and external tools to tell you what your cluster is doing -- in real-time. Read more…. Work-flows Simple and complex work-flows can be composed using a set of powerful primitives we call the "canvas", including grouping, chaining, chunking, and more. Read more…. Time & Rate Limits You can control how many tasks can be executed per second/minute/hour, or how long a task can be allowed to run, and this can be set as a default, for a specific worker or individually for each task type. Read more….

Scheduling You can specify the time to run a task in seconds or a datetime, or you can use periodic tasks for recurring events based on a simple interval, or Crontab expressions supporting minute, hour, day of week, day of month, and month of year. Read more…. Resource Leak Protection The --max-tasks-per-child option is used for user tasks leaking resources, like memory or file descriptors, that are simply out of your control. Read more…. User Components Each worker component can be customized, and additional components can be defined by the user. The worker is built up using "bootsteps" — a dependency graph enabling fine grained control of the worker's internals.

框架集成

Framework Integration

中文

Celery 易于与各类 Web 框架集成，其中一些框架甚至提供了专门的集成包：

Pyramid	https://pypi.org/project/pyramid_celery/
Pylons	https://pypi.org/project/celery-pylons/
Flask	不需要
web2py	https://pypi.org/project/web2py-celery/
Tornado	https://pypi.org/project/tornado-celery/
Tryton	https://pypi.org/project/celery_tryton/

关于 Django ，请参考 使用 Django 的第一步。

这些集成包并不是必需的， 但它们可以使开发过程更简单， 有时还提供了重要的扩展钩子， 例如在 fork(2) 时正确关闭数据库连接。

英文

Celery is easy to integrate with web frameworks, some of them even have integration packages:

Pyramid	https://pypi.org/project/pyramid_celery/
Pylons	https://pypi.org/project/celery-pylons/
Flask	not needed
web2py	https://pypi.org/project/web2py-celery/
Tornado	https://pypi.org/project/tornado-celery/
Tryton	https://pypi.org/project/celery_tryton/

For Django see 使用 Django 的第一步.

The integration packages aren't strictly necessary, but they can make development easier, and sometimes they add important hooks like closing database connections at fork(2).

快速跳转

Quick Jump

中文

英文

安装

Installation

中文

您可以通过 Python 包索引 (PyPI) 或从源代码安装 Celery。

使用 pip 安装：

英文

You can install Celery either via the Python Package Index (PyPI) or from source.

To install using pip:

$ pip install -U Celery

软件包

Bundles

中文

Celery 还定义了一组捆绑包（bundles）， 可用于安装 Celery 及特定功能所需的依赖项。

你可以在 requirements 文件中指定这些捆绑包， 或在 pip 命令行中通过使用中括号的方式指定。 若需指定多个捆绑包，可使用逗号分隔。

$ pip install "celery[librabbitmq]"

$ pip install "celery[librabbitmq,redis,auth,msgpack]"

可用的捆绑包包括：

英文

Celery also defines a group of bundles that can be used to install Celery and the dependencies for a given feature.

You can specify these in your requirements or on the pip command-line by using brackets. Multiple bundles can be specified by separating them by commas.

$ pip install "celery[librabbitmq]"

$ pip install "celery[librabbitmq,redis,auth,msgpack]"

The following bundles are available:

序列化器

Serializers

中文

celery[auth]:

用于启用 auth 安全序列化器。

celery[msgpack]:

用于启用 msgpack 序列化器。

celery[yaml]:

用于启用 yaml 序列化器。

英文

celery[auth]:

for using the auth security serializer.

celery[msgpack]:

for using the msgpack serializer.

celery[yaml]:

for using the yaml serializer.

并发

Concurrency

中文

celery[eventlet]:

用于启用 https://pypi.org/project/eventlet/ 池。

celery[gevent]:

用于启用 https://pypi.org/project/gevent/ 池。

英文

celery[eventlet]:

for using the https://pypi.org/project/eventlet/ pool.

celery[gevent]:

for using the https://pypi.org/project/gevent/ pool.

传输和后端

Transports and Backends

中文

celery[librabbitmq]:

用于启用 librabbitmq C 语言库。

celery[redis]:

用于将 Redis 用作消息传输或结果存储后端。

celery[sqs]:

用于将 Amazon SQS 用作消息传输（实验性）。

celery[tblib]:

用于启用 task_remote_tracebacks 功能。

celery[memcache]:

使用 Memcached 作为结果存储后端（基于 https://pypi.org/project/pylibmc/）。

celery[pymemcache]:

使用 Memcached 作为结果存储后端（纯 Python 实现）。

celery[cassandra]:

使用 Apache Cassandra / Astra DB 作为结果存储后端，基于 DataStax 驱动。

celery[couchbase]:

使用 Couchbase 作为结果存储后端。

celery[arangodb]:

使用 ArangoDB 作为结果存储后端。

celery[elasticsearch]:

使用 Elasticsearch 作为结果存储后端。

celery[riak]:

使用 Riak 作为结果存储后端。

celery[dynamodb]:

使用 AWS DynamoDB 作为结果存储后端。

celery[zookeeper]:

使用 Zookeeper 作为消息传输。

celery[sqlalchemy]:

使用 SQLAlchemy 作为结果存储后端（官方支持）。

celery[pyro]:

使用 Pyro4 作为消息传输（实验性）。

celery[slmq]:

使用 SoftLayer Message Queue 作为消息传输（实验性）。

celery[consul]:

使用 Consul.io Key/Value 存储作为消息传输或结果存储后端（实验性）。

celery[django]:

指定用于支持 Django 的最低版本。

通常不推荐在 requirements 中使用该选项， 这里仅作参考。

celery[gcs]:

使用 Google Cloud Storage 作为结果存储后端（实验性）。

celery[gcpubsub]:

使用 Google Cloud Pub/Sub 作为消息传输（实验性）。

英文

celery[librabbitmq]:

for using the librabbitmq C library.

celery[redis]:

for using Redis as a message transport or as a result backend.

celery[sqs]:

for using Amazon SQS as a message transport (experimental).

celery[tblib]:

for using the task_remote_tracebacks feature.

celery[memcache]:

for using Memcached as a result backend (using https://pypi.org/project/pylibmc/)

celery[pymemcache]:

for using Memcached as a result backend (pure-Python implementation).

celery[cassandra]:

for using Apache Cassandra/Astra DB as a result backend with DataStax driver.

celery[couchbase]:

for using Couchbase as a result backend.

celery[arangodb]:

for using ArangoDB as a result backend.

celery[elasticsearch]:

for using Elasticsearch as a result backend.

celery[riak]:

for using Riak as a result backend.

celery[dynamodb]:

for using AWS DynamoDB as a result backend.

celery[zookeeper]:

for using Zookeeper as a message transport.

celery[sqlalchemy]:

for using SQLAlchemy as a result backend (supported).

celery[pyro]:

for using the Pyro4 message transport (experimental).

celery[slmq]:

for using the SoftLayer Message Queue transport (experimental).

celery[consul]:

for using the Consul.io Key/Value store as a message transport or result backend (experimental).

celery[django]:

specifies the lowest version possible for Django support.

You should probably not use this in your requirements, it's here for informational purposes only.

celery[gcs]:

for using the Google Cloud Storage as a result backend (experimental).

celery[gcpubsub]:

for using the Google Cloud Pub/Sub as a message transport (experimental)..

从源代码下载并安装

Downloading and installing from source

中文

可以从 PyPI 下载 Celery 的最新版本：

https://pypi.org/project/celery/

你可以通过以下步骤进行安装：

$ tar xvfz celery-0.0.0.tar.gz
$ cd celery-0.0.0
$ python setup.py build
# python setup.py install

如果你未在虚拟环境（virtualenv）中， 最后一条命令需要以特权用户身份执行。

英文

Download the latest version of Celery from PyPI:

https://pypi.org/project/celery/

You can install it by doing the following,:

$ tar xvfz celery-0.0.0.tar.gz
$ cd celery-0.0.0
$ python setup.py build
# python setup.py install

The last command must be executed as a privileged user if you aren't currently using a virtualenv.

使用开发版本

Using the development version

使用 pip

With pip

中文

Celery 的开发版还需要依赖以下项目的开发版： https://pypi.org/project/kombu/ 、 https://pypi.org/project/amqp/ 、 https://pypi.org/project/billiard/ 和 https://pypi.org/project/vine/ 。

可以使用以下 pip 命令安装它们的最新快照版：

$ pip install https://github.com/celery/celery/zipball/main#egg=celery
$ pip install https://github.com/celery/billiard/zipball/main#egg=billiard
$ pip install https://github.com/celery/py-amqp/zipball/main#egg=amqp
$ pip install https://github.com/celery/kombu/zipball/main#egg=kombu
$ pip install https://github.com/celery/vine/zipball/main#egg=vine

英文

The Celery development version also requires the development versions of https://pypi.org/project/kombu/, https://pypi.org/project/amqp/, https://pypi.org/project/billiard/, and https://pypi.org/project/vine/.

You can install the latest snapshot of these using the following pip commands:

$ pip install https://github.com/celery/celery/zipball/main#egg=celery
$ pip install https://github.com/celery/billiard/zipball/main#egg=billiard
$ pip install https://github.com/celery/py-amqp/zipball/main#egg=amqp
$ pip install https://github.com/celery/kombu/zipball/main#egg=kombu
$ pip install https://github.com/celery/vine/zipball/main#egg=vine

使用 git

With git

中文

详情请参考 贡献指南。

英文

Please see the Contributing section.