芹菜

Question

现在，我想将< code>register事件发布到某个特殊的交换，我可以使用celery远程检索和处理它。

实际上，我已经使用了send_task函数来实现这一点，但是它必须传递task_name来指示应该执行哪个任务并消费它。所以它似乎不太适合我的目标。

我想要的就是这样:

1. 向某些Exchange发布注册消息；
2. 远程机器1订阅此主题或route_key并捕获消息，用于执行任务；
3. 远程机器2-与机器1相同但执行另一个任务-接收（可能需要回复某些队列）

例如，就像这个工作流一样：

注册：

>

send_email

generate_info

......

Answer 1

如果我需要非标准交换，我会这样做。

在我的celeryconfig中，我像这样指定该交换并将队列分配给它（在我的情况下，我需要fan out交换）：

from kombu.common import Broadcast
from kombu import Exchange, Queue
CELERY_QUEUES = (
   Broadcast(name='queue_name', exchange=Exchange('queue_name', type='fanout')),
)

然后，我用celery multi生成worker，并将其分配给我的特定队列，如下所示：

celery multi start 1 -A my_project -Q:1 queue_name -c:1 1 (other options go here)

然后我可以将我的任务插入该队列，如下所示：

from my_project import my_fancy_task
my_fancy_task.apply_async(args=(x, y, z), queue="my_queue")

我不太理解您的具体使用情况，如果您需要一台主机上的工作程序使用一个队列中的任务，然后另一台主机上的工作程序使用另一个队列中的任务，那么只需将您的任务分成两个队列，并配置每台主机来启动工作程序，然后将它们分配到对您有意义的队列中。也许这将有所帮助:用RabbitMQ交换模糊话题