鼠兔加芹菜，连接已关闭答案

【问题标题】：pika with celery, connection closed鼠兔加芹菜，连接已关闭
【发布时间】：2014-11-26 09:21:32
【问题描述】：

我正在使用 celery 和 RabbitMQ 来运行一些任务，有时我需要将工作人员的消息返回给 RabbitMQ，所以我使用 pika。

我目前正在使用 BlockingConnection() 来连接到 RabbitMQ，但过了一会儿我得到一个异常“连接丢失”。

我相信这是因为 celery 是异步的，而且我正在使用 BlockingConnection()。

这是我的代码：

class RabbitConnection(object):
    def __init__(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
        self.channel = self.connection.channel()
        self.channel.queue_declare(queue=RABBITMQ_OUT_NAME, durable=True)
        self.channel.confirm_delivery()

    def add_alert(self, new_alert):
        message = new_alert.to_json()
        delivered = self.channel.basic_publish(exchange='',
                                               routing_key=RABBITMQ_OUT_NAME,
                                               body=message,
                                               properties=pika.BasicProperties(
                                                   delivery_mode=2,
                                                   content_type='application/json',
                                               ))

我应该使用不同的连接吗？如果是，我应该如何使用它？

【问题讨论】：

这是多线程的吗？ Pika 被限制为每个线程一个连接。你可以试试我自己的线程安全库github.com/eandersson/amqp-storm
听起来不错，能否根据我的代码给我一个简单的示例，如何实现您的线程安全库？
我发布了一些示例代码。如果我错过了什么，请告诉我，因为我没有时间测试代码。
了解一下，3年左右。但是，连接关闭发生在我身上，因为它正在尝试使用 IPv6 地址。因此，我尝试了 127.0.0.1，它成功了。

标签： python rabbitmq pika

【解决方案1】：

听起来这可能是线程问题。您可以使用 Pika 在多个线程上处理请求，但理想情况下，每个线程应该有一个连接，或 use locking。我建议您使用线程安全库，而不是为您的代码增加额外的复杂性；比如amqp-storm或rabbitpy。

如果您使用我的 AMQP-Storm 库来实现此功能，代码将如下所示。

import amqpstorm

class RabbitConnection(object):
    def __init__(self):
        self.connection = amqpstorm.Connection('localhost', 'guest', 'guest')
        self.channel = self.connection.channel()
        self.channel.queue.declare(queue=RABBITMQ_OUT_NAME, durable=True)
        self.channel.confirm_deliveries()

    def add_alert(self, new_alert):
        message = new_alert.to_json()
        delivered = self.channel.basic.publish(exchange='',
                                               routing_key=RABBITMQ_OUT_NAME,
                                               body=message,
                                               properties={
                                                   'delivery_mode': 2,
                                                   'content_type': 'application/json',
                                                })

【讨论】：

这个例子是线程安全的还是我应该做其他事情？

【解决方案2】：

如果您愿意为每个请求牺牲约 5 毫秒的延迟：

import pika

class PikaClient:
    def __init__(self):
        self.connection = None
        self.channel = None

    def __enter__(self):
        connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
        channel = connection.channel()
        channel.queue_declare(queue='myqueue')
        self.connection = connection
        self.channel = channel
        return self

    def __exit__(self, type, value, traceback):
        self.connection.close()

    def messageSent(self, msgStr):
        self.channel.basic_publish(exchange='', routing_key='myqueue', body=msgStr)

然后当你想发送消息时：

with PikaClient() as pClient: 
    pClient.messageSent("my message")

根据您的应用程序，5 毫秒的延迟可能是值得付出的代价。如果您的应用程序成功，您可能无论如何都需要重写，然后您可以使用具有更好的多线程特性的语言，例如 Java。

【讨论】：