如何将芹菜任务作为在线程内运行的RabbitMQ使用者回调来执行？

Question

这个想法是： 我有一个kombu.mixins.ConsumerMixin对象在线程中运行。对于不同的主题，我需要不同的消费者，每个人执行不同的芹菜任务。因此，我想将一个芹菜任务传递到线程中，作为消费者回调。你知道吗

TopicConsumer类：

class TopicConsumer(ConsumerMixin):
    def __init__(self, connection, exchange_name, binding_keys):
        super().__init__()
        self.connection = connection
        self.exchange = Exchange(exchange_name, type='topic')

        self.q_bindings = [binding(self.exchange, routing_key=k) for k in binding_keys]
        self.q_name = f"{exchange_name}:{'|'.join(str(key) for key in binding_keys)}"
        self.queue = Queue(name=self.q_name,
                           exchange=self.exchange,
                           bindings=self.q_bindings)

    def get_consumers(self, Consumer, channel):
        consumers = list()
        consumer = Consumer(queues=self.queue,
                            accept=['json'],
                            callbacks=[self._handle_message])
        consumers.append(consumer)
        return consumers

    def _handle_message(self, body, message):
        try: 
            message.ack()
        except MessageStateError:
            # message has already been processed, rejected or requeued
            pass
        else:
            self.process_message(message.payload)

    def process_message(self, data):
        '''Must be overwritten by subclass.'''
        raise NotImplementedError('Subclass responsibility.')

消息内容如下：

import threading
from types import MethodType

from .TopicConsumer import TopicConsumer


class MessageConsumerThread(threading.Thread):
    def __init__(self, connection, exchange_name, binding_keys, msg_handler=None):
        super().__init__()
        self.connection = connection
        self.exchange_name = exchange_name
        self.binding_keys = binding_keys

        if msg_handler:
            self.register_message_handler(msg_handler)

    def register_message_handler(self, msg_handler):
        TopicConsumer.process_message = MethodType(msg_handler, TopicConsumer)
        self._create_consumer()

    def _create_consumer(self):
        self.consumer = TopicConsumer(self.connection, self.exchange_name, self.binding_keys)

    def run(self):
        self.consumer.run()

    def stop(self):
        self.consumer.should_stop = True

在主应用程序中有一个由函数调用的celery_task。你知道吗

def msg_handler(data):
    celery_task.delay(data)

我想使用types.MethodType注册msg_handler作为消息处理回调TopicConsumer.process_message()（我也尝试了setattr）。也不起作用。）

现在我试着这样做：

with Connection(**connection_parameters) as conn:
    msg_consumer = MessageConsumerThread(conn,
                                         exchange_name='messenger',
                                         binding_keys=['key'])
    msg_consumer.register_message_handler(msg_handler)
    msg_consumer.start()

如果像这样做，我可以从 MessageConsumerThread对象。但是它不是从TopicConsumer对象内部执行的，我需要它，因为有我想要处理的消息体。现在我知道如何让它运转了。你知道吗

你有什么建议吗？提前谢谢！你知道吗