python - 如何在运行时将队列添加到自定义消息使用者

Question

我有一个用例，我的 Django 应用程序需要处理来自外部源的消息。Celery 通过Custom Message Consumers支持这一点，这部分工作正常。

交换类型headers使得消息可以根据条件路由到不同的队列。条件是通过 Django Admin 配置的。从下面的代码中，您可以看到，例如 QueueItem 可以随时激活/停用。

这是第一次加载工人时发现的。在新的 Rabbit 上，交换、队列和路由都已构建，并且自定义消费者正确地使用消息。

问题出在用户添加新 QueueItem 或翻转现有 QueueItem 的活动状态时。我可以动态创建新队列，但无论我多么努力地尝试自定义消费者都不会从中消耗。我希望该应用程序：

1. 重新启动自定义消费者（我试过app.control.broadcast('pool_restart')）
2. 添加队列并使用自定义消费者使用它。（我试过app.control.add_consumer(...)）
3. 优雅地重新启动所有工作人员，当我手动工作以获取新设置时。

class MyConsumerStep(bootsteps.ConsumerStep):
    alias = 'MyConsumerStep'

    def get_queues(self):
        from app.models import QueueItem
        qs = QueueItem.objects.filter(active=True, mode=QueueItem.MODE_ASYNC, direction=QueueItem.DIRECTION_INBOUND)
        return [Queue(q.name, Exchange('qbpubsub', 'headers'),
                      binding_arguments={
                          'var1': q.field1,
                          'var2': q.field2,
                          'x-match': 'any'
                      }) for q in qs]

    def get_consumers(self, channel):
        return [Consumer(channel,
                         queues=self.service_queues,
                         on_message=self.on_message,
                         accept=['json'])]

    def on_message(self, message):
        payload = message.decode()
        print('Received message: {0!r} {props!r} rawlen={s}'.format(
            payload, props=message.properties, s=len(message.body)
        ))
        message.ack()

我已经成功地创建了一个新队列。我已经成功地将它附加到默认的 celery 消费者。我想以某种方式将它附加到自定义消费者，以便on_message回调可以处理它。

我觉得我已经很接近了，但只是缺少了 Celery / Kombu API 的一小部分。

标签： pythondjangopython-3.xcelerydjango-celery