如何在AMQP的python客户端中使用listen on basic.return

Question

我想确保我的消息已传递到队列中。

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

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

我使用 py-amqplib 版本 0.6。

欢迎任何解决方案。

Answer 1

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

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

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

由于对 pyAMQP 和rabbitMQ 的支持水平总体来说相当低，因此我们决定根本不使用 amqp 代理。

Answer 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

Answer 3

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

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

def registerCallback(channel, call_back):
    """ 此方法设置一个线程，用于处理无法由交换器路由的消息的异步回调。
    ”“”
    定义等待（）：
        尝试：
            通道.wait()
        除了异常，e：
            print("发布通道等待时出现问题：%s" % str(e))

        而不是channel.returned_messages.empty()：
            returnedMessage = Channel.returned_messages.get()
            processReturnedMessageThread = 线程(target=call_back, args=(returnedMessage))
            processReturnedMessageThread.start()

        等待（）

    等待=线程（目标=等待） 
    等待.start()