【发布时间】：2019-05-27 14:57:39
【问题描述】：
我正在使用 Rabbitmq 进行项目，我正在使用 RPC 模式，基本上我正在接收或使用来自队列的消息，进行一些处理，然后发送回响应。我正在使用 Pika，我的目标是为每个任务使用一个线程，因此对于每个任务，我都会特别为该任务创建一个线程。我还读到最好的做法是只建立一个连接，并在它下面有很多我想要的频道，但我总是得到这个错误：
'start_sumption 不能从'范围内调用'
pika.exceptions.RecursionError：不能从另一个 BlockingConnection 或 BlockingChannel 回调的范围调用 start_sumption。
我做了一些研究，发现 Pika 不是线程安全的，我们应该为每个线程使用一个独立的连接和一个通道。但我不想这样做，因为它被认为是不好的做法。所以我想在这里问一下是否有人已经完成了这项工作。我还读到如果我不使用 BlockingConnection 来实例化我的 Connection 是可能的，并且还有一个名为 add_callback_threadsafe 的函数可以使这成为可能。但不幸的是没有示例，我阅读了文档，但它很复杂，没有示例，我很难掌握他们想要描述的内容。
我的尝试是声明两个类。每个类将代表一个任务执行器，它接收或使用来自队列的消息，并基于该消息进行一些处理并返回响应。我的想法是在两个任务之间共享一个rabbitmq连接，但每个任务都会获得一个独立的频道。在上面的代码中，传递给函数的 rabbit 参数是一个类，它包含一些变量，如 Connection 和其他函数，如 EventSubscriber，当被调用时，它将分配一个新的 Channel 并开始使用来自特定 Exchanges 和 routingKey 的消息。接下来我声明一个线程并将订阅或消费函数作为该线程的目标。其他任务类看起来也和这个类一样，所以我只上传这个代码。在主类中，我与rabbitmq建立连接并将其作为参数传递给两个任务类的构造函数。
类 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")
主类：
类应用：
def __init__(self):
self.rabbitMq = RabbitMqCommunicationInterface(host='localhost', port=5672)
firstTask = On_Deregistration(self.rabbitMq)
secondTask = secondTask(self.rabbitMq)
app = App()
error : 'start_sumption 不能从'范围调用'
pika.exceptions.RecursionError: start_sumption 不能从另一个 BlockingConnection 或 BlockingChannel 回调的范围内调用
我搜索了这个错误的原因，显然 pika 不是线程安全的，但必须有一个解决方案。也许不使用 BlockingConnection ？也许有人可以给我一个如何做到这一点的示例，因为我尝试过但没有奏效。也许我错过了如何使用rabbitmq实现多线程
【问题讨论】：
标签： python multithreading rabbitmq pika