单连接、多通道的多线程 Kombu RPC 应用程序

Question

我有一个 RPC 应用程序，每个线程有一个连接和多个通道。目前，如果程序仅在单个线程中使用，则该程序可以按预期工作。然而，当添加多个线程时，会出现一系列意外问题，例如：

• amqp.exceptions.UnexpectedFrame：收到 0x31，同时期待 0xce
• 无法发送消息
• 收不到消息
• 通过对等点重置连接

RPC 应用程序的逻辑就在那里，因为它通过使用一个线程按预期工作，但使用多个线程是应用程序开始崩溃的地方。

我的应用程序的代码是：

from __future__ import absolute_import, unicode_literals

from kombu import Connection, Producer, Consumer, Queue, uuid
import uuid
import threading
import time
from utils import connector


class RpcConnection(object):
    def __init__(self):
        # connector variable is an amqp url to RabbitMq server
        self._connection = Connection(connector, heartbeat=10)
        self._pool = self._connection.ChannelPool(6)

    def connection(self):
        # Returns same connection
        return self._connection

    def new_channel(self):
        # Returns different channel each time from pool of channels
        return self._pool.acquire()


class ExecuteAction(object):
    def __init__(self, connection, channel):
        self._connection = connection
        self._channel = channel

        # Define single response queue on each instance of a thread
        self.callback_queue = Queue(name=str(uuid.uuid4()),
                                    exclusive=True,
                                    auto_delete=True,
                                    durable=False)

        self.correlation_id = None
        self.response = None

    def __exit__(self, exc_type, exc_val, exc_tb):
        self._channel.release()

    def execute_action(self):
        self.correlation_id = str(uuid.uuid4())

        # Send RPC message
        with Producer(self._channel) as producer:
            producer.publish(
                {'message_type': 'test',
                 'speed': int(1)},
                exchange='exchange.rpc',
                routing_key='message.action.rpc',
                reply_to=self.callback_queue.name,
                correlation_id=self.correlation_id,
            )

        # Consume RPC message back from queue
        with Consumer(self._channel,
                      on_message=self.on_response,
                      queues=[self.callback_queue],
                      no_ack=False):
            while self.response is None:
                print('Waiting for response')
                self._connection.drain_events()

        return self.response

    def on_response(self, message):
        if message.properties['correlation_id'] == self.correlation_id:
            print('Message received: {}\n'.format(message.payload))
            self.response = message.payload


class ThreadBase(threading.Thread):
    def __init__(self, connection, channel):
        threading.Thread.__init__(self)

        self._channel = channel
        self._connection = connection

        self.execute_action = None

    def initialise(self):
        # Initialises another class object which sends messages using RPC
        self.execute_action = ExecuteAction(connection=self._connection,
                                            channel=self._channel)

    def run(self):
        while True:
            # Infinite loop that calls execute_action() to send an RPC message every 4 seconds
            self.execute_action.execute_action()
            time.sleep(4)


if __name__ == '__main__':
    rpc_connection = RpcConnection()

    # For thread 1 and thread 2, the same connection is being passed, but each thread has a different channel
    # based on that same connection
    thread1 = ThreadBase(connection=rpc_connection.connection(),
                         channel=rpc_connection.new_channel())

    thread2 = ThreadBase(connection=rpc_connection.connection(),
                         channel=rpc_connection.new_channel())

    thread1.initialise()
    thread2.initialise()

    thread1.setName('Thread 1')
    thread2.setName('Thread 2')

    thread1.start()
    #time.sleep(2)
    thread2.start()

如果添加 2 秒或更多秒的

time.sleep()

，则应用程序可以按预期使用多个线程工作，但这不是期望的结果，因为线程彼此不并行工作。这里的目标是让多个线程通道使用相同的连接并行工作以发出和接收 RPC 调用。

非常感谢任何帮助。

Answer 1

您应该为每个线程使用一个连接，然后在该线程内创建通道。

另请注意，您的代码从两个单独的线程调用

self._connection.drain_events()

，但使用相同的连接。这肯定是有问题的。