如何在 AMQP 的 python 客户端中使用监听 basic.return答案

【问题标题】：How to use listen on basic.return in python client of AMQP如何在 AMQP 的 python 客户端中使用监听 basic.return
【发布时间】：2011-01-12 14:20:52
【问题描述】：

我想确保我的消息已发送到队列。

为此，我将强制参数添加到 basic_publish。如果我的消息没有成功送达，我还应该怎么做才能收到basic.return 消息？

我不能使用channel.wait() 来监听basic.return，因为当我的消息成功传递时，wait() 函数将永远挂起。（没有超时）另一方面。当我不呼叫channel.wait() 时，channel.returned_messages 将保持为空，即使消息未送达。

我使用py-amqplib 0.6 版。

欢迎任何解决方案。

【问题讨论】：

我对 python pika 库有完全相同的问题，似乎没有人谈论这个问题。我找不到任何演示如何处理失败消息的代码 sn-ps。

标签： python rabbitmq amqp py-amqplib

【解决方案1】：

目前不可能，因为当消息在代理中丢弃时，basic.return 是异步发送的。成功发送消息后，服务器不会报告任何数据。所以 pyAMQP 无法监听此类消息。

我已经阅读了一些关于这个问题的帖子。可能的解决方案是：

使用 txAMQP，处理 basic.return 的 amqp 的扭曲版本
使用带有超时等待的 pyAMQP。（我不确定目前是否可行）
使用同步命令频繁 ping 服务器，以便 pyAMQP 能够在 basic.return 消息到达时选择它们。

由于对 pyAMQP 和 rabbitMQ 的支持水平普遍很低，我们决定根本不使用 amqp 代理。

【讨论】：

【解决方案2】：

您是否尝试过唯一完整的 Python AMQP 库？它没有被广泛使用，因为它没有整齐地包装。

步骤 1. 编译 C 库 - 你可能需要sudo apt-get install autotools-dev autoconf automake libtool

mkdir rabbitc
cd rabbitc
hg clone http://hg.rabbitmq.com/rabbitmq-codegen/
hg clone http://hg.rabbitmq.com/rabbitmq-c/
cd rabbitmq-c
autoreconf -i
make clean
./configure --prefix=/usr
make
sudo make install

步骤 2. 安装 Python 库

pip install pylibrabbitmq

【讨论】：

【解决方案3】：

您不能同步执行此操作，因为它是一个异步系统。但是你可以使用线程来解决这个问题。

基本思想是，您启动一个线程，该线程在通道上进行等待，每当它退出等待时，它都会为返回的消息队列中的任何返回消息调用 call_back 函数。然后，您可以在 call_back 函数中处理该消息

def registerCallback(channel, call_back):
    """ This method sets up a thread which deals with the asynchronous callback for a message which could not be routed by the exchange.
    """
    def wait():
        try:
            channel.wait()
        except Exception, e:
            print("Problem waiting on publish channel: %s" % str(e))

        while not channel.returned_messages.empty():
            returnedMessage = channel.returned_messages.get()
            processReturnedMessageThread = Thread(target=call_back, args=(returnedMessage))
            processReturnedMessageThread.start()

        wait()

    waiting = Thread(target=wait) 
    waiting.start()

【讨论】：