处理 Celery 任务导致到达顺序

Question

我是 celery 新手，我有一个“celery-server”，就像下面的代码一样，它会在一段时间后返回结果，具体取决于计算。我在下面这个简单的程序中使用 sleep 功能模拟了这种行为。我想要的是在“重结果”之前处理早期返回的结果。我写了一个简单的程序，见下面的片段，它故意创建“重载”任务作为第一次调用。 请注意，随后的调用会创建“更轻”的任务，因此 celery 服务器会更早地返回它们。因此，我想根据它们到达客户端的顺序来处理返回的结果。现在（参见客户端代码）它一直等到繁重的任务返回。

但是对于 celery 文档中的示例，我应该通过检查 id 来等待结果，或者轮询它们（这很愚蠢，因为 celery 客户端必须以某种方式检查“第一个”到达结果的 id 我猜）。

如何按到达客户的顺序处理 celery 的结果？我不想在无休止的循环中轮询“result.ready()”，因为这完全搞砸了恕我直言，不知何故异步处理的感觉。

在文档中找不到解决方案。我想要做的是“获得第一个到达的结果并获得 id”，将其与我的“result.id”（我是否发送了任务？）进行比较，然后进行相应的处理。

#
# Name this code "tasks.py"  and run it with:
# celery worker -A tasks  --loglevel=info
#
from celery import Celery
import time
app = Celery('tasks', backend='amqp', broker='amqp://guest:guest@127.0.0.1:5672/%2F')

@app.task()
def add(x,y):
  print("x=%s y=%s" % (x,y))
  time.sleep(x)
  return x + y

对客户端进行第二次编程：这就像 celery 文档一样工作，但是 celery 已经完成了 0,1,2（因此客户端应该对其进行处理）。

#!/usr/bin/python3
from tasks import add

results = []
max = 4
for i in range(0,max):
  print(max-(i+1))
  result = add.delay(max-(i+1),0)
  results.append(result)

print("") 

for i in range(0,max):
  result = results[i].get(timeout=10)
  print(result)

结果：（最后 4 个数字应按到达顺序出现，即 0,1,2,3）

3
2
1
0

3
2
1
0

Answer 1

您应该实现一个回调，而不是按照发送到队列的顺序循环遍历结果：

http://celery.readthedocs.org/en/latest/userguide/calling.html#linking-callbacks-errbacks

在tasks.py中：

@app.task()
def process_add(result):
    print(result)

在client.py中：

from tasks import add, process_add
results = []
max = 4
for i in range(0,max):
    print(max-(i+1))
    add.apply_async((max-(i+1), 0), link=process_add.s())