5

我正在尝试用cellery进行申请。它应该适用于少数工作人员，并且不同的工作人员正在从不同的队列中消费。我有这样的事情：

@celery.task
def task1():
    do_something()
    task2.delay()

@celery.task
def task2()
    do_something()

所以在worker1上运行的task1应该调用task2，它应该从哪个消费worker2发送到队列。问题是它不起作用。我收到了 AsyncResult 的 id，但这个任务的状态一直是 PENDING。当我从 python 控制台手动调用 task2 时，它工作正常。也许我做错了什么，不可能从另一个任务中运行一个任务？还有一件事情。Worker1 正在执行 task1 并将 task2 发送到他不消费的队列 - 从这个队列中只消费 worker2

4

1 回答 1

0

这是一个简单的例子，我认为它可以完成你想要的。

from celery import Celery
import random
import string

celery = Celery('two_q',backend='amqp',broker='amqp://guest@localhost//')

@celery.task
def generate_rand_string(n):
    # n = number of characters
    rand_str = "".join([random.choice(string.lowercase) for i in range(n)])
    #calls the second task and adds it to second queue
    reverse.apply_async((rand_str,),queue="q2")
    print rand_str
    return rand_str

@celery.task
def reverse(s):
    print s[::-1]
    return s[::-1]

generate_rand_string.apply_async((10,), queue="q1")

当使用指定队列列表的 -Q 参数调用时

 celery worker --app=two_q -l info -Q q1,q2

它产生以下输出：

pawel@iqmxma82x7:~/py/celery$ celery worker --app=two_q -l info -Q q1,q2

 -------------- celery@iqmxma82x7 v3.0.23 (Chiastic Slide)
---- **** ----- 
--- * ***  * -- Linux-3.2.0-54-generic-pae-i686-with-Ubuntu-12.04-precise
-- * - **** --- 
- ** ---------- [config]
- ** ---------- .> broker:      amqp://guest@localhost:5672//
- ** ---------- .> app:         cel_group:0x9bfef8c
- ** ---------- .> concurrency: 4 (processes)
- *** --- * --- .> events:      OFF (enable -E to monitor this worker)
-- ******* ---- 
--- ***** ----- [queues]
 -------------- .> q1:          exchange:q1(direct) binding:q1
                .> q2:          exchange:q2(direct) binding:q2

[Tasks]
  . two_q.generate_rand_string
  . two_q.reverse

[2013-09-15 19:10:35,708: WARNING/MainProcess] celery@iqmxma82x7 ready.
[2013-09-15 19:10:35,716: INFO/MainProcess] consumer: Connected to amqp://guest@127.0.0.1:5672//.
[2013-09-15 19:10:40,731: INFO/MainProcess] Got task from broker: two_q.generate_rand_string[fa2ad56e-c66d-44a9-b908-2d95b2c9e5f3]
[2013-09-15 19:10:40,767: WARNING/PoolWorker-1] jjikjkepkc
[2013-09-15 19:10:40,768: INFO/MainProcess] Got task from broker: two_q.reverse[f52a8247-4674-4183-a826-d73cef1b64d4]
[2013-09-15 19:10:40,770: INFO/MainProcess] Task two_q.generate_rand_string[fa2ad56e-c66d-44a9-b908-2d95b2c9e5f3] succeeded in 0.0217289924622s: 'jjikjkepkc'
[2013-09-15 19:10:40,782: WARNING/PoolWorker-3] ckpekjkijj
[2013-09-15 19:10:40,801: INFO/MainProcess] Task two_q.reverse[f52a8247-4674-4183-a826-d73cef1b64d4] succeeded in 0.0195469856262s: 'ckpekjkijj'

你得到两个队列（q1，q2）和两个工人。

作为比较，如果您在没有 -Q 参数或仅使用一个队列的情况下调用它：

celery worker --app=two_q -l info

不会调用“反向”任务，因为 celery 不知道添加它的 q2。

希望能帮助到你。

于 2013-09-15T18:18:25.777 回答