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)
应用程序 = 应用程序()
pika.exceptions.RecursionError: start_sumption 不能从另一个 BlockingConnection 或 BlockingChannel 回调的范围内调用
我搜索了这个错误的原因,显然 pika 不是线程安全的,但必须有一个解决方案。也许不使用 BlockingConnection ?也许有人可以给我一个如何做到这一点的示例,因为我尝试过但没有奏效。也许我错过了一些关于如何使用 rabbitmq 实现多线程的信息
| 归档时间: |
|
| 查看次数: |
1464 次 |
| 最近记录: |