ConfirmListener.handleNack 缺少交换时不调用

Question

在我的应用程序中，我需要确定消息是否已成功发布到 AMQP 交换器或是否发生了某些错误。 Publisher Confirms 似乎是为了解决这个问题而发明的，所以我开始试验它们。

对于我的 Java 应用程序，我使用了 com.rabbitmq:amqp-client:jar:3.5.4 并且当交换（我尝试发布的地方）丢失时我选择了一个非常简单的场景。我预计 ConfirmListener.handleNack 会在这种情况下被调用。

这是我的 Java 代码：

package wheleph.rabbitmq_tutorial.confirmed_publishes;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ConfirmedPublisher {
    private static final Logger logger = LoggerFactory.getLogger(ConfirmedPublisher.class);

    private final static String EXCHANGE_NAME = "confirmed.publishes";

    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");

        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();

        channel.confirmSelect();
        channel.addConfirmListener(new ConfirmListener() {
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                logger.debug(String.format("Received ack for %d (multiple %b)", deliveryTag, multiple));
            }

            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                logger.debug(String.format("Received nack for %d (multiple %b)", deliveryTag, multiple));
            }
        });

        for (int i = 0; i < 100; i++) {
            String message = "Hello world" + channel.getNextPublishSeqNo();
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
            logger.info(" [x] Sent '" + message + "'");
            Thread.sleep(2000);
        }

        channel.close();
        connection.close();
    }
}

然而事实并非如此。日志显示没有执行回调：

17:49:34,988 [main] ConfirmedPublisher -  [x] Sent 'Hello world1'
Exception in thread "main" com.rabbitmq.client.AlreadyClosedException: channel is already closed due to channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'confirmed.publishes' in vhost '/', class-id=60, method-id=40)
    at com.rabbitmq.client.impl.AMQChannel.ensureIsOpen(AMQChannel.java:195)
    at com.rabbitmq.client.impl.AMQChannel.transmit(AMQChannel.java:309)
    at com.rabbitmq.client.impl.ChannelN.basicPublish(ChannelN.java:657)
    at com.rabbitmq.client.impl.ChannelN.basicPublish(ChannelN.java:640)
    at com.rabbitmq.client.impl.ChannelN.basicPublish(ChannelN.java:631)
    at wheleph.rabbitmq_tutorial.confirmed_publishes.ConfirmedPublisher.main(ConfirmedPublisher.java:38)

有趣的是，当我尝试使用 NodeJS 库 amqp-coffee (0.1.24) 时，发布者确认工作按预期进行。

这是我的 NodeJS 代码：

var AMQP = require('amqp-coffee');

var connection = new AMQP({host: 'localhost'});
connection.setMaxListeners(0);

console.log('Connection started')

connection.publish('node.confirm.publish', '', 'some message', {deliveryMode: 2, confirm: true}, function(err) {
     if (err && err.error && err.error.replyCode === 404) {
         console.log('Got 404 error')
     } else if (err) {
         console.log('Got some error')
     } else {
         console.log('Message successfully published')
     }
  })

这是指示使用正确参数调用回调的输出：

Connection started
Got 404 error

我使用 com.rabbitmq:amqp-client 不正确还是该库中存在一些不一致？

Answer 1

事实证明我的假设不正确，在这种情况下不应调用ConfirmListener.handleNack。

这是针对 amqp-coffee 库观察到的问题中描述的场景的 AMQP 消息的相关部分：

ch#1 -> {#method<channel.open>(out-of-band=), null, ""}
ch#1 <- {#method<channel.open-ok>(channel-id=), null, ""}
ch#1 -> {#method<confirm.select>(nowait=false), null, ""}
ch#1 <- {#method<confirm.select-ok>(), null, ""}
ch#1 -> {#method<basic.publish>(ticket=0, exchange=node.confirm.publish, routing-key=, mandatory=false, immediate=false), #contentHeader<basic>(content-type=string/utf8, content-encoding=null, headers=null, delivery-mode=2, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null), "some message"}
ch#1 <- {#method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'node.confirm.publish' in vhost '/', class-id=60, method-id=40), null, ""}
ch#2 -> {#method<channel.open>(out-of-band=), null, ""}
ch#2 <- {#method<channel.open-ok>(channel-id=), null, ""}
ch#2 -> {#method<confirm.select>(nowait=false), null, ""}
ch#2 <- {#method<confirm.select-ok>(), null, ""}

你可以看到：

发布不成功后，代理使用 channel.close 关闭了频道，其中包括原因。
basic.nack 未发送。
图书馆自动开启另一个通道进行后续操作

此行为可以在 Java 中使用 ShutdownListener 实现：

package wheleph.rabbitmq_tutorial.confirmed_publishes;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ShutdownListener;
import com.rabbitmq.client.ShutdownSignalException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ConfirmedPublisher {
    private static final Logger logger = LoggerFactory.getLogger(ConfirmedPublisher.class);
    private final static String EXCHANGE_NAME = "confirmed.publishes";

    // Beware that proper synchronization of channel is needed because current approach may lead to race conditions
    private volatile static Channel channel;

    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);

        final Connection connection = connectionFactory.newConnection();

        for (int i = 0; i < 100; i++) {
            if (channel == null) {
                createChannel(connection);
            }
            String message = "Hello world" + i;
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
            logger.info(" [x] Sent '" + message + "'");
            Thread.sleep(2000);
        }

        channel.close();
        connection.close();
    }

    private static void createChannel(final Connection connection) throws IOException {
        channel = connection.createChannel();
        channel.confirmSelect(); // This in fact is not necessary
        channel.addShutdownListener(new ShutdownListener() {
            public void shutdownCompleted(ShutdownSignalException cause) {
                // Beware that proper synchronization is needed here
                logger.debug("Handling channel shutdown...", cause);
                if (cause.isInitiatedByApplication()) {
                    logger.debug("Shutdown is initiated by application. Ignoring it.");
                } else {
                    logger.error("Shutdown is NOT initiated by application. Resetting the channel.");
                    /* We cannot re-initialize channel here directly because ShutdownListener callbacks run in the connection's thread,
                       so the call to createChannel causes a deadlock since it blocks waiting for a response (whilst the connection's thread
                       is stuck executing the listener). */
                    channel = null;
                }
            }
        });
    }
}

有几点注意事项：

在这种情况下不需要发布者确认，因为我们不使用 ConfirmListener 或任何其他特定于该方法的功能。但是，如果我们想跟踪哪些消息已成功发送，哪些未成功发送，则发布者确认会很有用。
如果我们启动 ConfirmedPublisher 并在一段时间后创建缺少的交换，则以下所有消息都将成功发布。但是所有以前失败的消息都丢失了。
需要额外的同步。

ConfirmListener.handleNack 缺少交换时不调用

ConfirmListener.handleNack is not invoked when exchange is missing

java

amqp

rabbitmq