Celery 4.3.0 中文文档：分布式任务队列详解

1星需积分: 0 134 浏览量更新于2024-07-15 收藏 18.66MB PDF 举报

"Celery-v4.3.0中文.pdf 是一份关于Celery分布式任务队列系统的中文文档，由AskSolem编著，发布于2019年3月31日。Celery是一个用于处理大量消息的简单、灵活、可靠的分布式系统，它专注于实时处理并支持任务调度。该文档涵盖了从入门到高级使用的各种内容，包括版权信息、入门指南、任务队列概念等。此外，文档还鼓励用户加入社区并为Celery的未来发展做出贡献。" ** Celery核心概念 ** 1. **任务队列**：Celery是一个基于消息的任务队列，任务是可异步执行的工作单元。任务被发送到队列，然后由工作进程（worker）取出并执行。这种机制允许将工作负载分散到多个处理器或服务器，提高系统的并发能力。 2. **分布式**：Celery设计用于分布式环境，可以跨越多个节点和服务器工作，这使得它可以处理大规模的计算任务，同时保持系统的稳定性和可靠性。 3. **实时处理与任务调度**：Celery不仅支持实时处理任务，还具有调度功能，可以设置定时任务或者延时任务，满足不同的业务需求。 4. **通信机制**：Celery通过消息传递来协调各个组件，常见的消息中间件如RabbitMQ、Redis或Amazon SQS等可以作为Celery的消息 broker，而结果后端则用于存储任务执行的结果。 ** 入门步骤 ** - **初学者教程**：对于不熟悉Celery的用户，可以按照文档中的“Celery的第一步”教程开始，逐步了解如何设置项目、定义任务、启动工作进程以及发送任务。 - **FAQ**：文档中可能包含一个FAQ（常见问题解答）部分，帮助解决在使用过程中遇到的问题。 ** 框架集成 ** Celery通常与其他Web框架如Django、Flask等集成，以扩展这些框架的功能，实现异步任务处理。在集成过程中，开发者可以利用Celery的API轻松地创建和管理任务。 ** 特性 ** - **灵活性**：Celery支持多种消息中间件，可以根据项目需求选择合适的broker和result backend。 - **可扩展性**：可以通过增加更多的工作进程来扩展处理能力。 - **监控与管理**：Celery提供了监控和管理工具，如flower，用于查看任务状态、跟踪任务执行情况等。 - **错误处理**： Celery支持任务重试、异常处理和回调机制，以增强任务的健壮性。 ** 安装 ** 安装Celery通常涉及以下步骤： 1. 安装Celery库：使用pip install celery 命令。 2. 配置消息中间件和结果后端。 3. 在项目中导入Celery并初始化实例。 4. 定义任务函数，并将其注册到Celery实例。 5. 启动工作进程。 ** 社区参与 ** Celery拥有活跃的社区，用户可以通过IRC或邮件列表与开发者和其他用户交流，甚至可以成为项目的贡献者或赞助者，共同推动Celery的改进和发展。这份文档详细介绍了Celery的各个方面，对于希望使用或深入了解Celery的开发者来说，是一份宝贵的参考资料。

---   --  [Configuration]  
*  ***    *
--  
-
-
-
-
- ---  .  broker:   amqp://guest@localhost:5672//  
__main__:0x1012d8590  
*
**  
**  
**  
**  
***  
****  
----------  .  app:  
----------  .  concurrency:  8  (processes)  
----------  .  events:  
----------  
OFF  (enable  -E  to  monitor  this  worker)   
- ---   ---  [Queues]  
*
--  
---  
----  .  celery:  
-----  
exchange:celery(direct)  binding:celery   
*******  
*****  
[2012-06-08  16:23:51,078:  WARNING/MainProcess]  celery@halcyon.local  has  started.   
3.2.  Getting Sta rted   29  
Celery Documentation, Release 4.3.0  
– The broker is the URL you speciﬁed in the broker argument in our celery module, you can also specify a different  
broker on the command-line by using the -b option.  
– Concurrency is the number of prefork worker process used to process your tasks concurrently, when all of these are  
busy doing work new tasks will have to wait for one of the tasks to ﬁnish before it can be processed.  
The default concurrency number is the number of CPU’s on that machine (including cores), you can specify a custom  
number using the celery  worker  -c option. There’s no recommended value, as the optimal number depends on  
a number of factors, but if your tasks are mostly I/O-bound then you can try to increase it, experimentation has shown  
that adding more than twice the number of CPU’s is rarely effective, and likely to degrade performance instead.  
Including the default prefork pool, Celery also supports using Eventlet, Gevent, and running in a single thread (see  
Concurrency).  
– Events is an option that when enabled causes Celery to send monitoring messages (events) for actions occurring  
monitor, that you can read about in the Monitoring and Management guide.  
– Queues is the list of queues that the worker will consume tasks from. The worker can be told to consume from several  
queues at once, and this is used to route messages to speciﬁc workers as a means for Quality of Service, separation of  
concerns, and prioritization, all described in the Routing Guide.  
$  celery  worker  --help   
These options are described in more detailed in the Workers Guide.  
Stopping the worker  
To stop the worker simply hit Control-c. A list of signals supported by the worker is detailed in the Workers Guide.  
In the background  
In production you’ll want to run the worker in the background, this is described in detail in the daemonization tutorial.  
The daemonization scripts uses the celery  multi command to start one or more workers in the background:  
$  celery  multi  start  w1  -A  proj  -l  info   
celery  multi  v4.0.0  (latentcall)   
>  Starting  nodes...   
>  w1.halcyon.local:  OK   
$  celery  multi  restart  w1  -A  proj  -l  info   
celery  multi  v4.0.0  (latentcall)   
>  Stopping  nodes...   
>  w1.halcyon.local:  TERM  ->  64024   
>  Waiting  for  1  node.....  
>  w1.halcyon.local:  OK   
>  Restarting  node  w1.halcyon.local:  OK   
celery  multi  v4.0.0  (latentcall)   
>  Stopping  nodes...   
>  w1.halcyon.local:  TERM  ->  64052   
or stop it:  
30   Chapter 3.  Contents  
in the worker.  These can be used by monitor programs like celery  events, and Flower - the real-time Celery  
You can get a complete list of command-line arguments by passing in the --help ﬂag:  
You can restart it too:  
Celery Documentation, Release 4.3.0  
$  celery  multi  stop  w1  -A  proj  -l  info   
stopwait command instead, this ensures all currently executing tasks are completed before exiting:  
$  celery  multi  stopwait  w1  -A  proj  -l  info   
Note:  celery  multi doesn’t store information about workers so you need to use the same command-line argu-  
ments when restarting. Only the same pidﬁle and logﬁle arguments must be used when stopping.  
By default it’ll create pid and log ﬁles in the current directory, to protect against multiple workers launching on top of  
each other you’re encouraged to put these in a dedicated directory:  
$  mkdir  -p  /var/run/celery   
$  mkdir  -p  /var/log/celery   
$  celery  multi  start  w1  -A  proj  -l  info  --pidfile=/var/run/celery/%n.pid  \   
--logfile=/var/log/celery/%n%I.log  
With the multi command  you  can start multiple workers, and there’s  a  powerful command-line syntax to specify  
arguments for different workers too, for example:  
$  celery  multi  start  10  -A  proj  -l  info  -Q:1-3  images,video  -Q:4,5  data  \   
-Q  default  -L:4,5  debug   
For more examples see the multi module in the API reference.  
About the --app argument  
The --app argument speciﬁes the Celery app instance to use, it must be in the form of module.path:attribute  
But it also supports a shortcut form If only a package name is speciﬁed, where it’ll try to search for the app instance,  
in the following order:  
With --app=proj:  
1)  an attribute named proj.app, or  
2)  an attribute named proj.celery, or  
3)  any attribute in the module proj where the value is a Celery application, or  
If none of these are found it’ll try a submodule named proj.celery:  
4)  an attribute named proj.celery.app, or  
5)  an attribute named proj.celery.celery, or  
6)  Any attribute in the module proj.celery where the value is a Celery application.  
This scheme mimics the practices used in the documentation – that is, proj:app for a single contained module, and  
proj.celery:app for larger projects.  
Calling Tasks  
 
The stop command is asynchronous so it won’t wait for the worker to shutdown.  You’ll probably want to use the  

3.2. Getting Sta rted 31

You can call a task using the delay() method:

Celery Documentation, Release 4.3.0

>>> add.delay(2, 2)

This method is actually a star-argument shortcut to another method called apply_async():

>>> add.apply_async((2, 2))

The latter enables you to specify execution options like the time to run (countdown), the queue it should be sent to,

and so on:

>>> add.apply_async((2, 2), queue='lopri', countdown=10)

In the above example the task will be sent to a queue named lopri and the task will execute, at the earliest, 10

seconds after the message was sent.

Applying the task directly will execute the task in the current process, so that no message is sent:

>>> add(2, 2)

These three methods - delay(), apply_async(), and applying (__call__), represents the Celery calling API,

that’s also used for signatures.

A more detailed overview of the Calling API can be found in the Calling User Guide.

Every task invocation will be given a unique identiﬁer (an UUID), this is the task id.

The delay and apply_async methods return an AsyncResult instance, that can be used to keep track of the

tasks execution state. But for this you need to enable a result backend so that the state can be stored somewhere.

Results are disabled by default because of the fact that there’s no result backend that suits every application, so to

choose one you need to consider the drawbacks of each individual backend. For many tasks keeping the return value

isn’t even very useful, so it’s a sensible default to have. Also note that result backends aren’t used for monitoring tasks

and workers, for that Celery uses dedicated event messages (see Monitoring and Management Guide).

If you have a result backend conﬁgured you can retrieve the return value of a task:

>>> res = add.delay(2, 2)

>>> res.get(timeout=1)

>>> res.id

d6b3aea2-fb9b-4ebc-8da4-848818db9114

gate any errors by default:

>>> res = add.delay(2)

>>> res.get(timeout=1)

Traceback (most recent call last):

File "<stdin>", line 1, in <module>

File "/opt/devel/celery/celery/result.py", line 113, in get

interval=interval)

File "/opt/devel/celery/celery/backends/rpc.py", line 138, in wait_for

raise meta['result']

TypeError: add() takes exactly 2 arguments (1 given)

32 Chapter 3. Contents

You can ﬁnd the task’s id by looking at the id attribute:

You can also inspect the exception and traceback if the task raised an exception, in fact result.get() will propa-

Celery Documentation, Release 4.3.0

If you don’t wish for the errors to propagate then you can disable that by passing the propagate argument:

>>> res.get(propagate=False)

TypeError('add() takes exactly 2 arguments (1 given)',)

In this case it’ll return the exception instance raised instead, and so to check whether the task succeeded or failed

you’ll have to use the corresponding methods on the result instance:

>>> res.failed()

True

>>> res.successful()

False

So how does it know if the task has failed or not? It can ﬁnd out by looking at the tasks state:

>>> res.state

'FAILURE'

A task can only be in a single state, but it can progress through several states. The stages of a typical task can be:

PENDING -> STARTED -> SUCCESS

The started state is a special state that’s only recorded if the task_track_started setting is enabled, or if the

@task(track_started=True) option is set for the task.

The pending state is actually not a recorded state, but rather the default state for any task id that’s unknown: this you

can see from this example:

>>> from proj.celery import app

>>> res = app.AsyncResult('this-id-does-not-exist')

>>> res.state

'PENDING'

If the task is retried the stages can become even more complex. To demonstrate, for a task that’s retried two times the

stages would be:

PENDING -> STARTED -> RETRY -> STARTED -> RETRY -> STARTED -> SUCCESS

To read more about task states you should see the States section in the tasks user guide.

Calling tasks is described in detail in the Calling Guide.

Canvas: Designing Work-ﬂows

may want to pass the signature of a task invocation to another process or as an argument to another function, for this

Celery uses something called signatures.

A signature wraps the arguments and execution options of a single task invocation in a way such that it can be passed

to functions or even serialized and sent across the wire.

>>> add.signature((2, 2), countdown=10)

tasks.add(2, 2)

3.2. Getting Sta rted 33

You just learned how to call a task using the tasks delay method, and this is often all you need, but sometimes you

You can create a signature for the add task using the arguments (2, 2), and a countdown of 10 seconds like this:

Celery Documentation, Release 4.3.0

There’s also a shortcut using star arguments:

>>> add.s(2, 2)

tasks.add(2, 2)

And there’s that calling API again. . .

Signature instances also supports the calling API: meaning they have the delay and apply_async methods.

But there’s a difference in that the signature may already have an argument signature speciﬁed. The add task takes

two arguments, so a signature specifying two arguments would make a complete signature:

>>> s1 = add.s(2, 2)

>>> res = s1.delay()

>>> res.get()

But, you can also make incomplete signatures to create what we call partials:

# incomplete partial: add(?, 2)

>>> s2 = add.s(2)

s2 is now a partial signature that needs another argument to be complete, and this can be resolved when calling the

signature:

# resolves the partial: add(8, 2)

>>> res = s2.delay(8)

>>> res.get()

Here you added the argument 8 that was prepended to the existing argument 2 forming a complete signature of add(8, 2).

Keyword arguments can also be added later, these are then merged with any existing keyword arguments, but with new

arguments taking precedence:

>>> s3 = add.s(2, 2, debug=True)

>>> s3.delay(debug=False) # debug is now False.

As stated signatures supports the calling API: meaning that;

• sig.apply_async(args=(), kwargs={},

options) **

Calls the signature with optional partial arguments and partial keyword arguments. Also supports

partial execution options.

• sig.delay( args, kwargs)

** *

Star argument version of apply_async. Any arguments will be prepended to the arguments in the signature,

and keyword arguments is merged with any existing keys.

So this all seems very useful, but what can you actually do with these? To get to that I must introduce the canvas

primitives. . .

The Primitives

34 Chapter 3. Contents

Celery Documentation, Release 4.3.0

• group

• chain

• chord

• map

• starmap

• chunks

These primitives are signature objects themselves, so they can be combined in any number of ways to compose

complex work-ﬂows.

Note: These examples retrieve results, so to try them out you need to conﬁgure a result backend. The example project

above already does that (see the backend argument to Celery).

Let’s look at some examples:

Groups

A group calls a list of tasks in parallel, and it returns a special result instance that lets you inspect the results as a

>>> from celery import group

>>> from proj.tasks import add

>>> group(add.s(i, i) for i in xrange(10))().get()

[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

• Partial group

>>> g = group(add.s(i) for i in xrange(10))

>>> g(10).get()

[10, 11, 12, 13, 14, 15, 16, 17, 18, 19]

Chains

Tasks can be linked together so that after one task returns the other is called:

>>> from celery import chain

>>> from proj.tasks import add, mul

# (4 + 4) 8*

>>> chain(add.s(4, 4) | mul.s(8))().get()

or a partial chain:

>>> # (? + 4)

>>> g = chain(add.s(4) | mul.s(8))

>>> g(4).get()

Chains can also be written like this:

3.2. Getting Started 35

group, and retrieve the return values in order.

Celery Documentation, Release 4.3.0

>>> (add s(4 4) | mul s(8))() get()

>>>  (add.s(4,  4)  |  mul.s(8))().get()  
64  
Chords  
A chord is a group with a callback:  
>>>  from  celery  import  chord  
>>>  from  proj.tasks  import  add,  xsum   
>>>  chord((add.s(i,  i)  for  i  in  xrange(10)),  xsum.s())().get()  
90  
A group chained to another task will be automatically converted to a chord:  
>>>  (group(add.s(i,  i)  for  i  in  xrange(10))  |  xsum.s())().get()  
90  
Since these primitives are all of the signature type they can be combined almost however you want, for example:  
>>>  upload_document.s(file)  |  group(apply_filter.s()  for  filter  in  filters)  
Be sure to read more about work-ﬂows in the Canvas user guide.  
Routing  
sent to named queues.  
The task_routes setting enables you to route tasks by name and keep everything centralized in one location:  
app.conf.update(  
task_routes  =  {   
'proj.tasks.add':  {'queue':  'hipri'},   
},  
)
>>>  from  proj.tasks  import  add  
>>>  add.apply_async((2,  2),  queue='hipri')  
$  celery  -A  proj  worker  -Q  hipri   
from both the default queue, and the hipri queue, where the default queue is named celery for historical reasons:  
$  celery  -A  proj  worker  -Q  hipri,celery   
The order of the queues doesn’t matter as the worker will give equal weight to the queues.  
To learn more about routing, including taking use of the full power of AMQP routing, see the Routing Guide.  
36   Chapter 3.  Contents  
 
Celery supports all of the routing facilities provided by AMQP, but it also supports simple routing where messages are  
You can also specify the queue at runtime with the queue argument to apply_async:  
You can then make a worker consume from this queue by specifying the celery  worker  -Q option:  
You may specify multiple queues by using a comma separated list, for example you can make the worker consume  
Celery Documentation, Release 4.3.0  
Remote Control  
If you’re using RabbitMQ (AMQP), Redis, or  Qpid as the  broker then you can control  and inspect the worker at   
runtime.  
For example you can see what tasks the worker is currently working on:  
$  celery  -A  proj  inspect  active   
This is implemented by using broadcast messaging, so all remote control commands are received by every worker in  
separated list of worker host names:  
$  celery-项目检查活动-目的地= celery@example.com  
如果未提供目的地，那么每个工作人员都会采取行动并回复请求。  
将芹菜检查命令包含没有在工人改变什么命令，它只是回复   
有关工人内部状况的信息和统计信息。有关检查命令的列表，您可以执行：  
$  celery-项目检查-帮助   
然后是celery控制命令，其中包含实际更改工作程序中的内容的命令   
在运行时：  
$  celery-项目控制-帮助   
例如，您可以强制工作程序启用事件消息（用于监视任务和工作程序）：  
$  celery-一个项目控制enable_events   
启用事件后，您可以启动事件转储程序以查看工作程序在做什么：  
$  celery  -A  proj  events  --dump   
或者您可以启动curses界面：  
$  芹菜项目   
完成监视后，可以再次禁用事件：  
$  celery-一个项目控件disable_events   
将芹菜状态命令也使用远程控制命令，并显示在集群中线上的工人名单：   
$  celery-项目状态   
时区  
内部和消息中的所有时间和日期都使用UTC时区。  
当工作人员收到一条消息（例如设置了倒计时）时，它将把UTC时间转换为本地时间。如果你  
希望使用与系统时区不同的时区，那么必须使用时区设置来配置它：  
3.2。入门指南    37  
 
 
the cluster.  
You can also specify one or more workers to act on the request using the --destination option. This is a comma  
Ÿ Ø u能阅读更多关于芹菜命令和监测监控指南。  
Celery文档，版本4.3.0  
应用程式。conf 。时区=  “欧洲/伦敦”  
优化  
默认情况下，默认配置并未针对吞吐量进行优化，而是尝试在许多  
短任务而长任务较少，这在吞吐量和合理调度之间进行了折衷。  
如果您有严格的公平调度要求 或者要针对吞吐量进行优化 则应阅读优化
 

如果您有严格的公平调度要求，或者要针对吞吐量进行优化，则应阅读优化  
指南。  
如果使用RabbitMQ，则可以安装librabbitmq模块：这是用C实现的AMQP客户端：  
$  pip安装librabbitmq   
现在做什么？  
现在，您已经阅读了本文档，应该继续阅读《用户指南》。  
如果您愿意的话，还有一个API参考。  
3.2.5资源   
•  获得帮助  
–  邮件列表  
–  IRC  
•  错误追踪器  
•  维基  
•  贡献  
•  执照  
获得帮助  
邮件列表  
For discussions about the usage, development, and future of Celery, please join the celery-users mailing list.  
IRC  
Come chat with us on IRC. The #celery channel is located at the Freenode network.  
Bug tracker  
If you have any suggestions, bug reports, or annoyances please report them to our issue tracker at https://github.com/  
celery/celery/issues/  
38   Chapter 3.  Contents  
 
 
 
 
 
Celery Documentation, Release 4.3.0  
Wiki  
https://wiki.github.com/celery/celery/  
Contributing  
Development of celery happens at GitHub: https://github.com/celery/celery  
you’re welcome to send regular patches.  
Be sure to also read the Contributing to Celery section in the documentation.  
License  
This software is licensed under the New BSD License. See the LICENSE ﬁle in the top distribution directory for the  
full license text.  
3.3  User Guide  
Release  4.3  
Date  Mar 31, 2019  
3.3.1  Application   
•  Main Name  
•  Conﬁguration  
•  Laziness  
•  Breaking the chain  
•  Abstract Tasks  
The Celery library must be instantiated before use, this instance is called an application (or app for short).  
The application is thread-safe so that multiple Celery applications with different conﬁgurations, components, and tasks  
can co-exist in the same process space.  
Let’s create one now:  
>>>  from  celery  import  Celery  
>>>  app  =  Celery()  
>>>  app  
<Celery  __main__:0x100469fd0>   
The last line shows the textual representation of the application: including the name of the app class (Celery), the  
name of the current main module (__main__), and the memory address of the object (0x100469fd0).  
3.3.  User Guide   39  
 
 
 
 
 
You’re highly encouraged to participate in the development of celery.  If you don’t like GitHub (for some reason)  
Celery Documentation, Release 4.3.0  
Main Name  
Only one of these is important, and that’s the main module name. Let’s look at why that is.  
When you send a task message in Celery, that message won’t contain any source code, but only the name of the task  
you want to execute. This works similarly to how host names work on the internet: every worker maintains a mapping  
of task names to their actual functions, called the task registry.  
Whenever you deﬁne a task, that task will also be added to the local registry:  
>>>  @app.task  
...  def  add(x,  y):   
...   return  x  +  y   
>>>  add  
<@task:  __main__.add>   
>>>  add.name  
main add
 