与Celery和RabbitMQ的主题交流

Question

我对设置主题交换的配置应该是什么样子有点困惑。

http://www.rabbitmq.com/tutorials/tutorial-five-python.html

这就是我想要完成的：

Task1 -> send to QueueOne and QueueFirehose
Task2 -> sent to QueueTwo and QueueFirehose

然后：

Task1 -> consume from QueueOne
Task2 -> consume from QueueTwo
TaskFirehose -> consume from QueueFirehose

我只希望Task1从QueueOne消费，Task2从QueueTwo消费。

现在的问题是，当Task1和Task2运行时，它们也会耗尽QueueFirehose，而TaskFirehose任务永远不会执行。

是我的配置有问题，还是我误解了什么？

CELERY_QUEUES = { 
    "QueueOne": {
        "exchange_type": "topic",
        "binding_key": "pipeline.one",
    },  
    "QueueTwo": {
        "exchange_type": "topic",
        "binding_key": "pipeline.two",
    },  
    "QueueFirehose": {
        "exchange_type": "topic",
        "binding_key": "pipeline.#",
    },  
}
CELERY_ROUTES = {
        "tasks.task1": {
            "queue": 'QueueOne',
            "routing_key": 'pipeline.one',
        },
        "tasks.task2": {
            "queue": 'QueueTwo',
            "routing_key": 'pipeline.two',
        },
        "tasks.firehose": {
            'queue': 'QueueFirehose',
            "routing_key": 'pipeline.#',
        },
}

Answer 1

假设你的意思是这样的：

Task1 -> send to QueueOne
Task2 -> sent to QueueTwo
TaskFirehose -> send to QueueFirehose

然后：

Worker1 -> consume from QueueOne, QueueFirehose
Worker2 -> consume from QueueTwo, QueueFirehose
WorkerFirehose -> consume from QueueFirehose

这可能不完全是你的意思，但我认为它应该涵盖许多场景，希望你也能。像这样的东西应该起作用：

# Advanced example starting 10 workers in the background:
#   * Three of the workers processes the images and video queue
#   * Two of the workers processes the data queue with loglevel DEBUG
#   * the rest processes the default' queue.
$ celery multi start 10 -l INFO -Q:1-3 images,video -Q:4,5 data
-Q default -L:4,5 DEBUG

有关更多选项和参考：http://celery.readthedocs.org/en/latest/reference/celery.bin.multi.html

这是直接从文档中得到的。

我也遇到过类似的情况，我处理的方式略有不同。我不能和主管一起用芹菜多糖。所以我在supervisord中为每个工人创建了多个程序。不管怎样，工人们会在不同的流程上工作，所以让主管为你处理一切。配置文件看起来像：-

; ==================================
; celery worker supervisor example
; ==================================
[program:Worker1]
; Set full path to celery program if using virtualenv
command=celery worker -A proj --loglevel=INFO -Q QueueOne, QueueFirehose
directory=/path/to/project
user=nobody
numprocs=1
stdout_logfile=/var/log/celery/worker1.log
stderr_logfile=/var/log/celery/worker1.log
autostart=true
autorestart=true
startsecs=10
; Need to wait for currently executing tasks to finish at shutdown.
; Increase this if you have very long running tasks.
stopwaitsecs = 600
; When resorting to send SIGKILL to the program to terminate it
; send SIGKILL to its whole process group instead,
; taking care of its children as well.
killasgroup=true
; if rabbitmq is supervised, set its priority higher
; so it starts first
priority=998

类似地，对于Worker2和WorkerFirehose，编辑相应的行以生成：

[program:Worker2]
; Set full path to celery program if using virtualenv
command=celery worker -A proj --loglevel=INFO -Q QueueTwo, QueueFirehose

和

[program:WorkerFirehose]
; Set full path to celery program if using virtualenv
command=celery worker -A proj --loglevel=INFO -Q QueueFirehose

将它们全部包含在supervisord.conf文件中，这样就可以了。