使用RPC后端获取芹菜任务结果

Question

我很难从芹菜的任务中得到结果。我的应用程序入口点如下所示：

from app import create_app,celery
celery.conf.task_default_queue = 'order_master'
order_app = create_app('../config.order_master.py')

现在，在启动应用程序之前，我先启动RabbitMQ并确保它没有队列：

root@3d2e6b124780:/# rabbitmqctl list_queues
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
root@3d2e6b124780:/#

现在我启动应用程序。在启动之后，我仍然看到RabbitMQ中没有队列。当我从应用程序jobs.add_together.delay(2, 3)启动任务时，我会得到任务ID：

ralfeus@web-2 /v/w/order (multiple-instances)> (order) curl localhost/test
{"result":"a2c07de4-f9f2-4b21-ae47-c6d92f2a7dfe"}
ralfeus@web-2 /v/w/order (multiple-instances)> (order)

在那一刻，我可以看到我的队列中有一个消息：

root@3d2e6b124780:/# rabbitmqctl list_queues
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
name    messages
dd65ba89-cce9-3e0b-8252-c2216912a910    0
order_master   1
root@3d2e6b124780:/# 

现在我开始从事芹菜工作：

ralfeus@web-2 /v/w/order (multiple-instances)>
/usr/virtualfish/order/bin/celery -A main_order_master:celery worker --loglevel=INFO -n order_master -Q order_master --concurrency 2
INFO:app:Blueprints are registered

 -------------- celery@order_master v5.0.0 (singularity)
--- ***** -----
-- ******* ---- Linux-5.4.0-51-generic-x86_64-with-glibc2.29 2020-10-22 16:38:56
- *** --- * ---
- ** ---------- [config]
- ** ---------- .> app:         app:0x7f374715c5b0
- ** ---------- .> transport:   amqp://guest:**@172.17.0.1:5672//
- ** ---------- .> results:     rpc://
- *** --- * --- .> concurrency: 2 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
 -------------- [queues]
                .> order_master            exchange=order_master(direct) key=order_master


[tasks]
  . app.jobs.add_together
  . app.jobs.post_purchase_orders

[2020-10-22 16:38:57,263: INFO/MainProcess] Connected to amqp://guest:**@172.17.0.1:5672//
[2020-10-22 16:38:57,304: INFO/MainProcess] mingle: searching for neighbors
[2020-10-22 16:38:58,354: INFO/MainProcess] mingle: all alone
[2020-10-22 16:38:58,375: INFO/MainProcess] celery@order_master ready.
[2020-10-22 16:38:58,377: INFO/MainProcess] Received task: app.jobs.add_together[f855bec7-307d-4570-ab04-3d036005a87b]
[2020-10-22 16:40:38,616: INFO/ForkPoolWorker-2] Task app.jobs.add_together[f855bec7-307d-4570-ab04-3d036005a87b] succeeded in 100.13561034202576s: 5

因此，它是可见的，工人可以捡起任务并执行它，并产生一个结果。但是我无法得到结果。相反，当我请求结果时，我得到以下结果：

curl localhost/test/f855bec7-307d-4570-ab04-3d036005a87b
{"state":"PENDING"}
ralfeus@web-2 /v/w/order (multiple-instance)> (order)

如果我现在检查队列，我会发现：

root@3d2e6b124780:/# rabbitmqctl list_queues
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
name    messages
dd65ba89-cce9-3e0b-8252-c2216912a910    1
65d80661-6195-3986-9fa2-e468eaab656e    0
celeryev.9ca5a092-9a0c-4bd5-935b-f5690cf9665b   0
order_master   0
celery@order_master.celery.pidbox      0
root@3d2e6b124780:/#

我看到队列dd65ba89-cce9-3e0b-8252-c2216912a910有一条消息，在我检查包含结果时。那么为什么它出现在那里，我怎么得到它呢？所有的手册都说我只需要通过ID获取任务，但在我的例子中，任务仍然处于待定状态。

Answer 1

根据芹菜文件：

RPC结果后端(RabbitMQ/QPid)

RPC结果后端(rpc://)是特殊的，因为它实际上并不存储状态，而是将它们作为消息发送。这是一个重要的区别，因为它意味着结果只能检索一次，而只能由发起任务的客户端检索。两个不同的进程不能等待相同的结果。

因此，使用rpc://不适合稍后通过另一个请求检索结果。