如何使用 pika 和 rabbitmq 的多线程来执行请求和响应 RPC 消息

bas*_*isk 1 python multithreading rabbitmq pika

我正在使用 Rabbitmq 进行一个项目,我正在使用 RPC 模式,基本上我正在接收或使用来自队列的消息,进行一些处理,然后发送回响应。我使用 Pika,我的目标是为每个任务使用一个线程,因此对于每个任务,我将专门为该任务创建一个线程。我还读到,最佳实践是只建立一个连接,并根据需要在其下建立多个通道,但我总是收到此错误:
“可能无法从 pika.exceptions.RecursionError 的范围内调用 start_sumption”:start_sumption 可能不会从另一个 BlockingConnection 或 BlockingChannel 回调的范围内调用。

我做了一些研究,发现 Pika 不是线程安全的,我们应该为每个线程使用一个独立的连接和一个通道。但我不想这样做,因为它被认为是不好的做法。所以我想在这里问一下是否有人已经实现了这项工作。我还读到,如果我不使用 BlockingConnection 来实例化我的 Connection 并且有一个名为 add_callback_threadsafe 的函数可以使这成为可能。但不幸的是没有示例,我阅读了文档,但它很复杂,如果没有示例,我很难理解他们想要描述的内容。

我的尝试是声明两个类。每个类将代表一个任务执行器,它接收或使用来自队列的消息,并基于此进行一些处理并返回响应。我的想法是在两个任务之间共享一个rabbitmq 连接,但每个任务都会获得一个独立的通道。在上面的代码中,传递给函数的 rabbit 参数是一个类,其中包含一些变量,如 Connection 和其他函数,如 EventSubscriber,当调用它时,它将分配一个新通道并开始使用来自该特定交换和路由密钥的消息。接下来我声明一个线程并将订阅或消费函数作为该线程的目标。另一个任务类看起来也和这个类一样,这就是为什么我只上传这个代码。

On_Deregistration 类:

def __init__(self, rabbit):
   self.event(rabbit) # this will call event function and pass the connection shared between all Tasks. rabbit parameter hold a connection to rabbitmq

def event(self, rabbit):
 
    self.Subscriber = rabbit.EventSubscriber(rabbit,  'testing.test',  'test', False,  onDeregistrationFromHRS # this func is task listener)

def subscribeAsync(self):
    self.Subscriber.subscribe() # here i call start_consuming

def start(self):
    """start Subscribtion in an Independant Thread  """
    thread = threading.Thread(target = self.subscribeAsync )  
    thread.start()
    if thread.isAlive():
        print("asynchronous subscription started")
Run Code Online (Sandbox Code Playgroud)

主课:

类应用程序:

def __init__(self):

    self.rabbitMq = RabbitMqCommunicationInterface(host='localhost', port=5672)
    firstTask =  On_Deregistration(self.rabbitMq)
    secondTask =  secondTask(self.rabbitMq)
Run Code Online (Sandbox Code Playgroud)

应用程序 = 应用程序()

错误:'start_sumption 可能无法从 ' 的范围内调用

pika.exceptions.RecursionError: start_sumption 不能从另一个 BlockingConnection 或 BlockingChannel 回调的范围内调用

我搜索了这个错误的原因,显然 pika 不是线程安全的,但必须有一个解决方案。也许不使用 BlockingConnection ?也许有人可以给我一个如何做到这一点的示例,因为我尝试过但没有奏效。也许我错过了一些关于如何使用 rabbitmq 实现多线程的信息

bas*_*isk 5

所以经过长时间的研究,我发现 Pika 不是线程安全的。至少就目前而言,也许在新版本中它将是线程安全的。所以现在对于我的项目,我停止使用 Pika 而我使用的是b-rabbit,它是 Rabbitpy 周围的线程安全包装器。但我必须说 Pika 是一个很棒的库,我发现 API 的描述和结构比 rabbitpy 更好,但是对于我的项目来说,必须使用多线程,这就是为什么 Pika 目前是一个糟糕的选择。我希望这对未来的人有所帮助