java - 当交换丢失时，不会调用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, ""}

你可以看到:

1. 发布失败后，代理会使用 channel.close 关闭 channel ，并说明原因。
2. basic.nack 未发送。
3. 库会自动打开另一个 channel 进行后续操作。

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

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;
                }
            }
        });
    }
}

有一些注意事项:

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