我有一个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调用。
任何帮助都非常感谢。