celery官方文档笔记

2023-12-16 13:01:21

启动和后台运行

启动celery worker server

linux

celery -A yiyan worker -l info

windows

`celery -A yiyan worker -l info -E -P eventlet`

`-E是启用flower监控`
`-P是windows下运行必备的参数`

启动flower，默认端口5555

celery -A yiyan flower

获取命令帮助

celery worker --help

celery --help

Celery Worker运行在后台

详细信息可以在“daemonization tutorial”（守护进程教程）中找到

启动一个或多个后台工作者：

celery multi start w1 -A proj -l INFO

重启

celery multi restart w1 -A proj -l INFO

停止

celery multi stop w1 -A proj -l INFO

请注意，stop 命令是异步的，它不会等待工作者完全关闭。相反，你可能想要使用 stopwait 命令，它确保在退出之前完成所有当前正在执行的任务：

celery multi stopwait w1 -A proj -l INFO

需要注意的是，celery multi 不会存储有关工作者的信息，因此在重新启动时需要使用相同的命令行参数。在停止时，必须使用相同的 pidfile 和 logfile 参数。

默认情况下，celery multi 会在当前目录中创建 pid 和 log 文件。为了防止多个工作者在彼此之上启动，建议将它们放在专用目录中：

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

使用 multi 命令，你可以启动多个工作者，并且有一个强大的命令行语法，可以为不同的工作者指定参数，例如：

celery multi start 10 -A proj -l INFO -Q:1-3 images,video -Q:4,5 data -Q default -L:4,5 debug

有关更多示例，请参阅 API 参考中的 multi 模块。

任务

from myapp.tasks import add
result = add.delay(1, 4)
print("任务ID: ",result.id)
print("任务结果: ",result.get(result.id))  # 阻塞，等待任务完成，一旦这样用，就会把异步调用变成了同步调用
print("检查任务是否处理完成: ",result.ready())
# PENDING -> STARTED -> RETRY -> STARTED -> RETRY -> STARTED -> SUCCESS
print("检查任务状态: ",result.state)
print("检查任务是否失败: ",result.failed())
print("检查任务是否成功: ",result.successful())

# from celery import group
# r = group(add.s(i, i) for i in range(10))()
# print(r)
# print(r.get())

多个装饰器在与任务装饰器一起使用多个装饰器时，你必须确保任务装饰器最后应用（奇怪的是，在 Python 中这意味着它必须是列表中的第一个）：

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

绑定任务任务绑定意味着任务的第一个参数将始终是任务实例（self），就像 Python 中的绑定方法一样：

logger = get_task_logger(__name__)

@app.task(bind=True)
def add(self, x, y):
    logger.info(self.request.id)

绑定任务对于重试（使用 app.Task.retry()）、访问有关当前任务请求的信息以及添加到自定义任务基类的任何其他功能都是必需的。

调用任务时传递参数和禁用参数检查

Celery在调用任务时会验证传递的参数，就像在调用普通函数时Python所做的那样：

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

# 调用任务传递两个参数是可以的：
add.delay(8, 8)
<AsyncResult: f59d71ca-1549-43e0-be41-4e8821a83c0c>

# 调用任务只传递一个参数会失败：
add.delay(8)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "celery/app/task.py", line 376, in delay
    return self.apply_async(args, kwargs)
  File "celery/app/task.py", line 485, in apply_async
    check_arguments(*(args or ()), **(kwargs or {}))
TypeError: add() takes exactly 2 arguments (1 given)

你可以通过将任务的typing属性设置为False来禁用任何任务的参数检查：

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

# 在本地运行是可以的，但接收任务的worker会引发错误。
add.delay(8)
<AsyncResult: f59d71ca-1549-43e0-be41-4e8821a83c0c>

任务请求对象

通常绑定后使用

任务请求 app.Task.request 包含与当前执行的任务相关的信息和状态。

请求定义了以下属性：

id: 执行任务的唯一标识。
group: 任务组的唯一标识，如果此任务是组的成员。
chord: 此任务所属的 chord 的唯一标识（如果任务是 header 的一部分）。
correlation_id: 用于诸如去重之类的事物的自定义 ID。
args: 位置参数。
kwargs: 关键字参数。
origin: 发送此任务的主机的名称。
retries: 当前任务已经重试的次数。从 0 开始的整数。
is_eager: 如果任务在客户端本地执行而不是由工作程序执行，则设置为 True。
eta: 任务的原始 ETA（如果有的话）。这是以 UTC 时间表示的（取决于 enable_utc 设置）。
expires: 任务的原始到期时间（如果有的话）。这是以 UTC 时间表示的（取决于 enable_utc 设置）。
hostname: 执行任务的工作程序实例的节点名称。
delivery_info: 附加的消息传递信息。这是一个映射，包含用于传递此任务的交换和路由键。例如，app.Task.retry() 用于将任务重新发送到相同的目标队列。此字典中的键的可用性取决于所使用的消息代理。
reply-to: 发送回复的队列的名称（例如，与 RPC 结果后端一起使用）。
called_directly: 如果任务不是由工作程序执行，则将此标志设置为 True。
timelimit: 当前任务的活动（软限制，硬限制）时间限制的元组（如果有的话）。
callbacks: 如果此任务成功返回，则将调用此任务的签名的列表。
errbacks: 如果此任务失败，则将调用此任务的签名的列表。
utc: 如果调用者启用了 UTC（enable_utc）则设置为 True。

从版本 3.1 开始。

headers: 与此任务消息一起发送的消息头的映射（可能为 None）。
reply_to: 用于发送回复的位置（队列名称）。
correlation_id: 通常与任务 ID 相同，经常在 amqp 中用于跟踪回复的对象。

从版本 4.0 开始。

root_id: 工作流程中此任务所属的第一个任务的唯一标识（如果有的话）。
parent_id: 调用此任务的任务的唯一标识（如果有的话）。

从版本 5.2 开始。

chain: 形成链的任务的反向列表（如果有的话）。此列表中的最后一项将是下一个成功执行当前任务的任务。如果使用任务协议的第一个版本，则链式任务将在 request.callbacks 中。
properties: 与此任务消息一起接收的消息属性的映射（可能为 None 或 {}）。
replaced_task_nesting: 任务替换了多少次，如果有的话（可能为 0）。

示例访问上下文中的信息的示例任务是：

@app.task(bind=True)
def dump_context(self, x, y):
    print('Executing task id {0.id}, args: {0.args!r} kwargs: {0.kwargs!r}'.format(self.request))

bind 参数意味着该函数将是一个“绑定方法”，因此你可以访问任务类型实例上的属性和方法。

任务状态

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

状态可推断，比如，任务的当前状态是failure或者success，那么他在之前的某一个时段肯定是经过started状态

为了确保资源被释放，你必须最终对调用任务后返回的每个AsyncResult实例调用get()或forget()。

内建状态（Built-in States）

指 Celery 任务在其生命周期中可能经历的状态。以下是一些内建状态及其说明：

PENDING：
- 状态说明： 任务正在等待执行或处于未知状态。任何未知的任务 id 都被隐式认为处于等待执行状态。
- 备注： 默认情况下不报告 STARTED 状态，要启用，请参阅 app.Task.track_started。
STARTED：
- 状态说明： 任务已经启动。
- 元数据： 包含执行任务的 worker 进程的 pid 和主机名。
SUCCESS：
- 状态说明： 任务已成功执行。
- 元数据： result 包含任务的返回值。
- 传播： 是
- 准备就绪： 是
FAILURE：
- 状态说明： 任务执行失败。
- 元数据： result 包含发生的异常，traceback 包含在引发异常时堆栈的回溯。
- 传播： 是
RETRY：
- 状态说明： 任务正在重试。
- 元数据： result 包含导致重试的异常，traceback 包含在引发异常时堆栈的回溯。
- 传播： 否
REVOKED：
- 状态说明： 任务已被撤销。
- 传播： 是

这些状态反映了任务在执行过程中可能的各种情况，例如任务成功执行、任务失败、任务重试等。这些状态信息对于监控和管理任务的执行非常有用。

Handlers

Celery提供了一组处理程序，你可以使用它们来连接到任务执行生命周期的不同阶段。以下是每个处理程序的说明：

before_start:
- 目的： 在任务开始执行之前由worker运行。
- 参数：
  - task_id：要执行的任务的唯一标识符。
  - args：任务执行的原始参数。
  - kwargs：任务执行的原始关键字参数。
- 返回值： 被忽略。
after_return:
- 目的： 在任务返回后调用的处理程序。
- 参数：
  - status：当前任务状态。
  - retval：任务返回值/异常。
  - task_id：任务的唯一标识符。
  - args：任务返回的原始参数。
  - kwargs：任务返回的原始关键字参数。
  - einfo：ExceptionInfo实例，包含回溯信息（如果有）。
- 返回值： 被忽略。
on_failure:
- 目的： 在任务失败时由worker运行。
- 参数：
  - exc：任务引发的异常。
  - task_id：失败任务的唯一标识符。
  - args：任务失败的原始参数。
  - kwargs：任务失败的原始关键字参数。
  - einfo：ExceptionInfo实例，包含回溯信息。
- 返回值： 被忽略。
on_retry:
- 目的： 在任务重试时由worker运行。
- 参数：
  - exc：传递给retry()的异常。
  - task_id：要重试的任务的唯一标识符。
  - args：要重试的任务的原始参数。
  - kwargs：要重试的任务的原始关键字参数。
  - einfo：ExceptionInfo实例，包含回溯信息。
- 返回值： 被忽略。
on_success:
- 目的： 如果任务成功执行，则由worker运行。
- 参数：
  - retval：任务的返回值。
  - task_id：执行的任务的唯一标识符。
  - args：执行的任务的原始参数。
  - kwargs：执行的任务的原始关键字参数。
- 返回值： 被忽略。

这些处理程序允许你在任务执行过程的不同阶段执行自定义操作或记录。例如，你可能希望在任务开始之前记录信息，以特定的方式处理任务失败的错误，或者在任务成功返回后执行清理任务。

需要注意的地方是Django模型对象。它们不应该作为参数传递给任务。最好的做法几乎总是在任务运行时从数据库中重新获取对象，因为使用旧数据可能导致竞态条件

想象一下以下场景，您有一篇文章和一个任务，该任务会自动扩展其中的一些缩写：

class Article(models.Model):
    title = models.CharField()
    body = models.TextField()

@app.task
def expand_abbreviations(article):
    article.body.replace('MyCorp', 'My Corporation')
    article.save()

首先，作者创建一篇文章并保存它，然后作者点击一个按钮，启动缩写任务：

article = Article.objects.get(id=102)
expand_abbreviations.delay(article)

现在，队列非常繁忙，因此任务将在另外2分钟后才能运行。与此同时，另一位作者对文章进行了更改，因此当任务最终运行时，文章的正文将恢复到旧版本，因为任务在其参数中具有旧的正文。

修复这个竞态条件很容易，只需改用文章ID，并在任务正文中重新获取文章：

@app.task
def expand_abbreviations(article_id):
    article = Article.objects.get(id=article_id)
    article.body.replace('MyCorp', 'My Corporation')
    article.save()

这种方法甚至可能带来性能上的好处，因为发送大型消息可能会很昂贵。

另一个数据库事务的例子

from django.db import transaction
from django.http import HttpResponseRedirect

@transaction.atomic
def create_article(request):
    article = Article.objects.create()
    expand_abbreviations.delay(article.pk)
    return HttpResponseRedirect('/articles/')

这是一个Django视图，在数据库中创建文章对象，然后将主键传递给任务。它使用transaction.atomic装饰器，该装饰器将在视图返回时提交事务，或者如果视图引发异常则回滚事务。

如果任务在事务提交之前开始执行，将存在竞态条件；数据库对象尚不存在！

解决方案是使用on_commit回调，在所有事务成功提交后启动Celery任务。

from django.db import transaction
from django.http import HttpResponseRedirect

@transaction.atomic
def create_article(request):
    article = Article.objects.create()
    transaction.on_commit(lambda: expand_abbreviations.delay(article.pk))  # 这很重要
    return HttpResponseRedirect('/articles/')

注意 on_commit在Django 1.9及以上版本中可用，如果您使用的是比这更早的版本，则django-transaction-hooks库会添加对此的支持。

调用任务

https://docs.celeryq.dev/en/v5.3.5/userguide/calling.html#results-options

Canvas设计工作流

签名

signature 函数是一个用于以灵活和可重用的方式构建任务调用的强大工具。它允许您封装任务应该如何被调用的细节，包括其参数、关键字参数和执行选项。

以下是关于使用 signature 的一些关键点和示例：

创建 Signature：

您可以使用任务的名称以及提供参数和选项来创建一个 signature。

示例：

from celery import signature

# 使用任务名称以及提供参数和选项
sig = signature('tasks.add', args=(2, 2), countdown=10)

或者，您可以使用任务的 signature 方法。

示例：

# 使用任务的 signature 方法
sig = tasks.add.signature((2, 2), countdown=10)

还有一种使用任务的 s 方法的快捷方式。

示例：

# 使用任务的 s 方法（快捷方式）
sig = tasks.add.s(2, 2)

检查 Signature：

一旦您有了一个 signature 实例，就可以检查其组件。

示例：

# 检查 signature 组件
sig.args  # 返回 (2, 2)
sig.kwargs  # 返回 {'debug': True}
sig.options  # 返回 {'countdown': 10}

调用 Signature：
- 可以直接调用 signature 以在当前进程中执行任务。
- 示例：
```
# 调用 signature
result = sig()
```
- 您还可以使用 delay 方法，这是 apply_async 方法的快捷方式，采用星号参数。
- 示例：
```
# 使用 delay 作为快捷方式
result = sig.delay()
```

异步应用：

apply_async 方法在任务和 signature 上都可用。

示例：

# 直接在任务上使用 apply_async
result = tasks.add.apply_async((2, 2), countdown=1)

# 通过 signature 使用 apply_async
result = sig.apply_async()

链接 Set：
- 不能使用 s() 定义选项，但可以使用链接的 set 调用来添加选项。
- 示例：
```
# 链接 set 添加选项
chained_sig = tasks.add.s(2, 2).set(countdown=1)
```

总之，signature 提供了一种清晰而富有表达力的方式来定义任务调用，使得更容易通过不同的参数和选项重用和组合任务调用。它是创建灵活且模块化任务配置的宝贵工具。

Partials部分调用

signature 不仅可以用于执行任务，还可以创建部分调用（partials）。部分调用是指对任务调用的一部分进行了设置，而返回一个新的 signature，使得您可以在稍后的时间或不同的上下文中使用它。

部分调用的创建：

添加额外参数、关键字参数或选项：

克隆 Signature：

Immutability不可变性

Callbacks 回调函数

回调函数通常用于在父任务完成后执行附加操作，这些操作可能依赖于父任务的结果。
回调函数只有在父任务成功完成时才会被调用，并且父任务的返回值会作为参数传递给回调函数。

Primitives原语（构建工作流）

group（组）：

概述： group 是一个签名，它接受一个任务列表，这些任务应该并行执行。
示例： 例如，您可以使用 group 将多个任务放在一组，以便它们可以并行执行。
```
group(task1.s(), task2.s(), task3.s()).apply_async()
```

chain （链）：

概述： chain 允许将多个签名链接在一起，以便一个任务在另一个任务之后被调用，形成一个回调链。
示例： 通过 chain，您可以按照顺序链接多个任务，确保它们按照期望的次序执行。
```
chain(task1.s(), task2.s(), task3.s()).apply_async()
```

不要在任务里面调用子任务，可以用任务链，通过链式调用创建任务链

def update_page_info(url):
    # fetch_page -> parse_page -> store_page
    chain = fetch_page.s(url) | parse_page.s() | store_page_info.s(url)
    chain()
	# 上面2行等同于my_chain = chain(fetch_page.s(url), parse_page.s(), store_page_info.s(url))

@app.task()
def fetch_page(url):
    return myhttplib.get(url)

@app.task()
def parse_page(page):
    return myparser.parse_document(page)

@app.task(ignore_result=True)
def store_page_info(info, url):
    PageInfo.objects.create(url=url, info=info)

以上是一个任务链的例子，这里有两个知识点：

顺序执行

A|B|C，按顺序执行，B接收A的结果作为输入，C接收B的结果作为输入，这个效果是由管道符|来实现的，|是链式调用的语法糖
A没执行完，不会执行B，严格按照A->B->C
调用chain()时，启动整个链开始执行。

异步调用

chain = fetch_page.s(url) | parse_page.s() | store_page_info.s(url) 创建了一个任务链，里面有三个带签名的任务，他们的执行顺序是fetch_page -> parse_page -> store_page，此时他们只是被创建出来放在了任务链里面，并不会被执行。
chain() 被调用，但它并不会等待整个链执行完毕。相反，它将整个任务链提交到 Celery 的任务队列中，然后立即返回。异步就体现在这。

整个链的执行是异步的，因为 chain() 的调用不会等待任务链的完成。程序可以继续执行后续的代码，而不必等待任务链中的每个任务执行完毕。

任务链中的每个任务的执行仍然是同步的，即一个任务的执行会等待前一个任务完成。但整个任务链的提交是异步的，它允许你在提交任务链后继续执行其他操作。异步提交是为了提高程序的并发性和响应性。

chord （和弦）：

概述： 和弦与组类似，但有一个回调函数。和弦由一个头部组和一个主体组成，主体是一个任务，在头部组的所有任务完成后执行。
示例： 使用和弦可以在一组任务完成后执行一个额外的任务，形成一种任务组合。

chord([task1.s(), task2.s(), task3.s()], task_after_group.s()).apply_async()

map（映射）：

概述： map 的工作方式类似于内置的 map 函数，但创建一个临时任务，其中一系列参数被应用于任务。
示例： 例如，使用 map 可以将一组参数依次应用于任务函数，形成一个结果列表。
```
task.map([1, 2])  # 结果为 [task(1), task(2)] 
```

starmap（星映射）：

概述： 与 map 完全相同，但参数作为 *args 应用。
示例： 使用 starmap 可以将一组参数作为 *args 应用于任务函数，形成一个结果列表。
```
add.starmap([(2, 2), (4, 4)])  # 结果为 [add(2, 2), add(4, 4)] 
```

chunks（块）：

概述： 块操作将长列表的参数分割成多个部分。例如，将一个包含1000个项目的列表分割成大小为10的块，将生成100个任务，每个任务依次处理10个项目。

示例：

items = zip(range(1000), range(1000))  # 1000个项目
add.chunks(items, 10)

这些原语提供了在Celery中组织和管理任务的有效方法，使得可以更容易地定义和执行任务的组合。

Workers

启动

在同一台机器上启动多个 Celery worker，并通过指定 --hostname 参数为每个工作进程命名。以下是相关内容的解释：

启动多个 Celery Worker

可以通过以下命令在同一台机器上启动多个 Celery worker：

bashCopy codecelery -A proj worker --loglevel=INFO --concurrency=10 -n worker1@%h
celery -A proj worker --loglevel=INFO --concurrency=10 -n worker2@%h
celery -A proj worker --loglevel=INFO --concurrency=10 -n worker3@%h

每个命令通过 -n 参数指定了一个不同的节点名，分别为 worker1@%h、worker2@%h 和 worker3@%h。其中 %h 是一个变量，将根据当前主机名进行展开。

节点名中的变量

--hostname 参数中的变量可以使用以下模板：

%h：主机名，包括域名。
%n：仅主机名。
%d：仅域名。

例如，如果当前主机名为 george.example.com，那么这些变量将展开为：

%h：worker1@george.example.com
%n：worker1@george
%d：worker1@example.com

注意事项

对于使用 Supervisor 的用户，百分号 % 必须通过添加第二个百分号进行转义，即 %%h。这是因为 Supervisor 本身可能会使用百分号作为特殊字符。

这样的设置允许您在同一台机器上运行多个 Celery worker，并通过节点名区分它们，确保它们能够独立运行。

停止

优雅关闭 Worker

要优雅地关闭 Celery worker 进程，应该使用 TERM 信号。在关闭时，worker 会在实际终止之前完成所有当前正在执行的任务。如果这些任务很重要，你应该等待它完成，然后再执行任何严重的操作，比如发送 KILL 信号。

强制终止 Worker

如果 worker 无法在合理的时间内关闭，可能因为它陷入无限循环或类似的问题，你可以使用 KILL 信号来强制终止 worker。但要注意，当前正在执行的任务将会丢失（除非任务设置了 acks_late 选项）。

由于进程无法覆盖 KILL 信号，worker 将无法回收其子进程，因此请确保手动执行此操作。以下命令通常可以完成这个任务：

pkill -9 -f 'celery'

如果你的系统上没有 pkill 命令，你可以使用稍微冗长的版本：

ps auxww | awk '/celery worker/ {print $2}' | xargs kill -9

注意事项

在 Celery 5.2 版本中，对于 Linux 系统，Celery 现在支持在 worker 终止后向所有子进程发送 KILL 信号。这是通过 prctl(2) 的 PR_SET_PDEATHSIG 选项完成的。

这些步骤确保了以一种安全的方式关闭 Celery worker，避免数据丢失和其他潜在问题。

重启

使用 celery multi（推荐）

在开发环境中，你可以使用 celery multi 来管理 workers。下面是一些示例命令：

启动一个 worker 实例：

celery multi start 1 -A proj -l INFO -c4 --pidfile=/var/run/celery/%n.pid

重新启动一个 worker 实例：

celery multi restart 1 --pidfile=/var/run/celery/%n.pid

这种方式简单易用，特别适用于开发环境。

使用 TERM 信号（适用于生产环境）

在生产环境中，你可以发送 TERM 信号给 worker，然后启动一个新的实例。这可以通过以下步骤完成：

发送 TERM 信号给 worker：
```
kill -TERM $pid
```
启动新的 worker 实例。

使用 HUP 信号（不推荐在生产环境使用）

你也可以使用 HUP 信号来重新启动 worker。这样的做法是给 worker 发送 HUP 信号，使其自行重启。但请注意，在生产环境中这可能会导致一些问题，不推荐使用。命令如下：

kill -HUP $pid

注意：

使用 HUP 信号重启只在 worker 以后台守护进程方式运行时有效（没有控制终端）。
在 macOS 上，由于平台限制，HUP 信号被禁用。

总体来说，在生产环境中，建议使用第一种或第二种方式，即 celery multi 或 TERM 信号。这些方式更为稳妥，易于集成到生产环境的管理系统中。

自动重连

启动失败自动重试一次broker_connection_retry_on_startup:

默认情况下，如果 broker_connection_retry_on_startup 被设置为 True，Celery 将在启动时首次连接消息代理失败后自动尝试重新连接到消息代理。

后续链接消息代理失败后重连broker_connection_retry:

broker_connection_retry 控制是否在后续重连时自动尝试重新连接到消息代理。
如果设置为 True，Celery 将在与消息代理连接丢失后自动尝试重新连接。

后续连接续链接消息代理失败后取消正在长时间运行的任务worker_cancel_long_running_tasks_on_connection_loss:

如果设置为 True，Celery 将在与消息代理连接丢失时取消任何当前正在运行的长时间运行的任务。
这意味着当与消息代理的连接丢失时，正在运行的长时间运行的任务将被取消。

后续连接续链接消息代理失败后重新获取任务的策略worker_prefetch_multiplier:

当与消息代理的连接丢失时，由于消息代理无法追踪在连接丢失前已经获取了多少任务，Celery 将通过减去当前正在运行的任务数量乘以 worker_prefetch_multiplier 来降低预取计数。
随着每个在连接丢失前正在运行的任务完成，预取计数将逐渐恢复到允许的最大值。

worker进程信号

TERM:
- 优雅关闭，等待任务完成后再终止。
QUIT:
- 冷关闭，尽快终止。
USR1:
- 为所有活动线程生成回溯(traceback)。
USR2:
- 远程调试，参见 celery.contrib.rdb。

此外，关于文件路径的一些参数，比如 --logfile、--pidfile 和 --statedb，可以包含一些变量，这些变量将由 worker 进程进行展开：

节点名称替换：

%p: 完整节点名称。
%h: 主机名，包括域名。
%n: 仅主机名。
%d: 仅域名。
%i: 预分叉池进程索引，如果是 MainProcess 则为 0。
%I: 预分叉池进程索引，带有分隔符。

例如，如果当前主机名是 george@foo.example.com，那么这些将被展开为：

--logfile=%p.log -> george@foo.example.com.log
--logfile=%h.log -> foo.example.com.log
--logfile=%n.log -> george.log
--logfile=%d.log -> example.com.log

这使得你可以根据实际的主机和配置信息来动态生成文件路径。

预分叉池进程索引Prefork pool process index

%i - 预分叉池进程索引：
- %i 表示预分叉池进程索引，或者在 MainProcess 时为 0。
- 使用 %i 可以根据进程的索引为每个子进程指定不同的文件名，以实现每个子进程有其独立的日志文件，适用于指定一个日志文件给每个子进程。
示例：
```
bashCopy code-n worker1@example.com -c2 -f %n-%i.log
```
这将生成三个日志文件：
- worker1-0.log（主进程）
- worker1-1.log（池进程 1）
- worker1-2.log（池进程 2）
%I - 带有分隔符的预分叉池进程索引：
- %I 表示带有分隔符的预分叉池进程索引。
- 使用 %I 可以根据进程的索引为每个子进程指定不同的文件名，适用于指定一个日志文件给每个子进程。
示例：
```
bashCopy code-n worker1@example.com -c2 -f %n%I.log
```
这将生成三个日志文件：
- worker1.log（主进程）
- worker1-1.log（池进程 1）
- worker1-2.log（池进程 2）

这样的配置使得你可以为每个池子进程指定不同的文件名，以更好地组织和区分日志输出。

并发性的配置

默认并发方式：
- 默认情况下，Celery 使用 multiprocessing 进行任务的并发执行。你也可以选择使用 Eventlet。
进程/线程数量配置：
- 通过 --concurrency 参数可以修改工作进程/线程的数量，默认值是机器上可用的 CPU 数量。
- 例如，--concurrency=4 表示使用 4 个工作进程或线程。
进程数量和性能：
- 通常来说，拥有更多的池子进程通常是更好的，但是存在一个临界点，当增加更多的池子进程时，性能会受到负面影响。
- 有证据表明，运行多个 worker 实例可能比运行单个 worker 更好，例如，3 个 worker 每个有 10 个池子进程。
- 你需要根据应用程序、工作负载、任务运行时间等因素进行实验，找到最适合你的数字。

远程控制

支持的并发方式：
- Celery 支持多种并发方式，如 prefork、eventlet、gevent、thread 以及 blocking:solo（注意：solo 并发池支持远程控制命令，但任何正在执行的任务都会阻塞任何等待的控制命令，因此在工作进程非常繁忙时，其用途有限）。
支持的消息代理：
- Celery 的远程控制功能支持多种消息代理，包括 amqp 和 redis。
远程控制命令：
- Celery worker 具有通过高优先级广播消息队列进行远程控制的能力。
- 命令可以发送到所有工作进程或指定列表中的特定工作进程。
- 命令还可以有回复，客户端可以等待并收集这些回复。
- 由于没有中央控制机构知道集群中有多少工作进程，因此无法估计可能发送回复的工作进程数量，客户端有一个可配置的超时时间，用于等待回复的截止时间（以秒为单位）。如果工作进程在截止时间内未回复，并不一定意味着工作进程没有回复，或者更糟糕的是已经死亡，而可能仅仅是由于网络延迟或工作进程处理命令的速度较慢，因此可以相应地调整超时时间。
- 除了超时时间，客户端还可以指定等待的最大回复数。如果指定了目标，此限制将设置为目标主机的数量。

总的来说，这使得你能够通过命令行向 Celery worker 发送控制命令，并在需要时等待回复。

广播函数

在 Celery 中，broadcast() 函数用于向工作进程发送控制命令。以下是一些关于使用 broadcast() 函数的示例：

发送 rate_limit 命令：

使用 broadcast() 函数发送 rate_limit 命令，并指定任务名称和速率限制。

app.control.broadcast('rate_limit',
                      arguments={'task_name': 'myapp.mytask',
                                 'rate_limit': '200/m'})

发送 rate_limit 命令并等待回复：
- 如果需要等待回复，可以使用 reply=True 参数。
```
app.control.broadcast('rate_limit', {
    'task_name': 'myapp.mytask', 'rate_limit': '200/m'}, reply=True)
```
- 这将返回一个包含每个工作进程回复的列表。

指定目标工作进程：

使用 destination 参数可以指定接收命令的工作进程列表。

app.control.broadcast('rate_limit', {
    'task_name': 'myapp.mytask',
    'rate_limit': '200/m'}, reply=True,
                         destination=['worker1@example.com'])

这将仅向指定的工作进程发送命令并等待回复。

总的来说，broadcast() 函数是用于向 Celery worker 发送控制命令的强大工具，通过它你可以实现对工作进程的灵活控制。

周期性任务

signals（信号）

在 Celery 中，信号是一种机制，用于在应用程序中的某些操作发生时通知其他部分。Celery 提供了许多信号，你的应用程序可以连接到这些信号以在某些操作发生时执行额外的操作。

以下是 Celery 中一些常见的信号及其基本用法：

Task Signals（任务信号）

before_task_publish: 在任务发布之前触发。
after_task_publish: 在任务发布之后触发。
task_prerun: 在任务运行之前触发。
task_postrun: 在任务运行之后触发。
task_retry: 在任务重试时触发。
task_success: 在任务成功完成时触发。
task_failure: 在任务失败时触发。
task_internal_error: 在任务内部错误时触发。
task_received: 在任务接收时触发。
task_revoked: 在任务撤销时触发。
task_unknown: 在任务未知状态时触发。
task_rejected: 在任务被拒绝时触发。

App Signals（应用程序信号）

import_modules: 在导入模块时触发。

Worker Signals（工作节点信号）

celeryd_after_setup: 在 Celery worker 启动后触发。
celeryd_init: 在 Celery worker 初始化时触发。
worker_init: 在 worker 进程初始化时触发。
worker_before_create_process: 在创建 worker 进程之前触发。
worker_ready: 在 worker 准备好接受任务时触发。
heartbeat_sent: 在发送心跳时触发。
worker_shutting_down: 在 worker 关闭时触发。
worker_process_init: 在 worker 进程初始化时触发。
worker_process_shutdown: 在 worker 进程关闭时触发。
worker_shutdown: 在 worker 关闭时触发。

Beat Signals（Beat 调度器信号）

beat_init: 在 Beat 初始化时触发。
beat_embedded_init: 在嵌入式 Beat 初始化时触发。

Eventlet Signals（Eventlet 池信号）

eventlet_pool_started: 在 Eventlet 池启动时触发。
eventlet_pool_preshutdown: 在 Eventlet 池关闭之前触发。
eventlet_pool_postshutdown: 在 Eventlet 池关闭之后触发。
eventlet_pool_apply: 在 Eventlet 池应用时触发。

Logging Signals（日志记录信号）

setup_logging: 在设置日志记录时触发。
after_setup_logger: 在设置完日志记录后触发。
after_setup_task_logger: 在设置完任务日志记录后触发。

Command signals（命令信号）

user_preload_options: 在预加载用户选项时触发。

Deprecated Signals（已弃用信号）

task_sent: 在任务发送时触发（已弃用）。

你可以连接到这些信号并定义相应的处理函数，以在特定事件发生时执行自定义操作。

----------------------------------------------------------

路由

task_routes 设置使您能够通过名称路由任务，并将所有内容集中在一个位置：

app.conf.update(
    task_routes = {
        'proj.tasks.add': {'queue': 'hipri'},
    },
)

您还可以在运行时使用 queue 参数指定队列到 apply_async：

from proj.tasks import add
add.apply_async((2, 2), queue='hipri')

然后，您可以通过指定 celery worker -Q 选项，使工作进程从这个队列中消耗：

celery -A proj worker -Q hipri

您可以通过使用逗号分隔的列表指定多个队列。例如，您可以使工作进程同时从默认队列和 hipri 队列消耗，其中默认队列出于历史原因被命名为 celery：

celery -A proj worker -Q hipri,celery

队列的顺序无关紧要，因为工作进程将对队列给予相同的权重。详见 http://docs.celeryproject.org/en/latest/userguide/routing.html

远程控制

关于Celery的远程控制，如果你使用RabbitMQ（AMQP）、Redis或者Qpid作为消息代理，你可以在运行时对worker进行控制和检查。

例如，你可以查看worker当前正在执行的任务：

celery -A proj inspect active

这是通过使用广播消息实现的，因此所有远程控制命令都会被集群中的每个worker接收。

你还可以使用 --destination 选项指定一个或多个要执行请求的worker。这是一个逗号分隔的worker主机名列表：

celery -A proj inspect active --destination=celery@example.com

如果没有提供目标，则每个worker都将执行并回复请求。

celery inspect 命令包含一些不会改变worker内部状态的命令，它仅返回有关worker内部情况的信息和统计信息。要查看所有inspect命令的列表，可以执行：

celery -A proj inspect --help

此外，celery control 命令包含一些在运行时实际更改worker状态的命令：

celery -A proj control --help

例如，你可以强制worker启用事件消息（用于监视任务和worker）：

celery -A proj control enable_events

启用事件后，你可以启动事件dumper查看worker的活动：

celery -A proj events --dump

或者你可以启动curses界面：

celery -A proj events

在监视完成后，你可以再次禁用事件：

celery -A proj control disable_events

celery status 命令也使用远程控制命令，并显示集群中在线的worker列表：

celery -A proj status

更多关于 celery 命令和监视的信息，请参阅 Monitoring Guide。

文章来源:https://blog.csdn.net/kpc0564335/article/details/135028355
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系我的编程经验分享网邮箱：veading@qq.com进行投诉反馈，一经查实，立即删除！

celery官方文档笔记

启动和后台运行

启动celery worker server

启动flower，默认端口5555

Celery Worker运行在后台

任务

任务

调用任务时传递参数和禁用参数检查

任务请求对象

任务状态

内建状态（Built-in States）

Handlers

需要注意的地方是Django模型对象。它们不应该作为参数传递给任务。最好的做法几乎总是在任务运行时从数据库中重新获取对象，因为使用旧数据可能导致竞态条件

另一个数据库事务的例子

调用任务

Canvas设计工作流

签名

Partials部分调用

Immutability不可变性

Callbacks 回调函数

Primitives原语（构建工作流）

group**（组）：**

chain （链）：

chord （和弦）：

map**（映射）：**

starmap**（星映射）：**

chunks**（块）：**

Workers

启动

启动多个 Celery Worker

节点名中的变量

注意事项

停止

优雅关闭 Worker

强制终止 Worker

注意事项

重启

使用 celery multi（推荐）

使用 TERM 信号（适用于生产环境）

使用 HUP 信号（不推荐在生产环境使用）

自动重连

启动失败自动重试一次broker_connection_retry_on_startup:

后续链接消息代理失败后重连broker_connection_retry:

后续连接续链接消息代理失败后取消正在长时间运行的任务worker_cancel_long_running_tasks_on_connection_loss:

后续连接续链接消息代理失败后重新获取任务的策略worker_prefetch_multiplier:

worker进程信号

预分叉池进程索引Prefork pool process index

并发性的配置

远程控制

广播函数

周期性任务

signals（信号）

Task Signals（任务信号）

App Signals（应用程序信号）

Worker Signals（工作节点信号）

Beat Signals（Beat 调度器信号）

Eventlet Signals（Eventlet 池信号）

Logging Signals（日志记录信号）

Command signals（命令信号）

Deprecated Signals（已弃用信号）

----------------------------------------------------------

路由

远程控制

group（组）：

map（映射）：

starmap（星映射）：

chunks（块）：