将 Celery 与现有的 RabbitMQ 消息一起使用

Question

我有一个现有的 RabbitMQ 部署，其中一些 Java 应用程序在各种通道上使用作为字符串 JSON 对象发送的日志消息。我想使用 Celery 来消费这些消息并将它们写入各个地方（例如 DB、Hadoop 等）。

我可以看到 Celery 被设计为既是 RabbitMQ 消息的生产者又是消费者，因为它试图隐藏传递这些消息的机制。有没有办法让 Celery 使用另一个应用程序创建的消息并在它们到达时运行作业？

Answer 1

目前很难将自定义消费者添加到 celery 工作者中，但这在开发版本（成为 3.1）中发生了变化，我添加了对消费者引导步骤的支持。

目前还没有文档，因为我刚刚完成了它，但这里有一个例子：

from celery import Celery
from celery.bin import Option
from celery.bootsteps import ConsumerStep
from kombu import Consumer, Exchange, Queue

class CustomConsumer(ConsumerStep):
   queue = Queue('custom', Exchange('custom'), routing_key='custom')

   def __init__(self, c, enable_custom_consumer=False, **kwargs):
       self.enable = self.enable_custom_consumer

   def get_consumers(self, connection):
       return [
           Consumer(connection.channel(),
               queues=[self.queue],
               callbacks=[self.on_message]),
       ]

   def on_message(self, body, message):
       print('GOT MESSAGE: %r' % (body, ))
       message.ack()


celery = Celery(broker='amqp://localhost//')
celery.steps['consumer'].add(CustomConsumer)
celery.user_options['worker'].add(
    Option('--enable-custom-consumer', action='store_true',
           help='Enable our custom consumer.'),
)

请注意，最终版本中的 API 可能会发生变化，我还不确定一件事 about 是在get_consumer(connection) 之后如何处理频道。 目前消费者的通道在连接丢失时关闭，在关闭时， 但人们可能想要手动处理频道。在那种情况下，总是有可能 自定义 ConsumerStep，或编写新的 StartStopStep。