RabbitMQ的消息确认机制

简介

RabbitMQ中提供了消息确认的机制，通过消息确认机制可以确保我们的消息可靠的送达到用户的手中。即便消息丢失掉，我们也可以通过重发来确保消息送达。
RabbitMQ提供了两种消息确认的机制：

通过AMQP事务机制来实现消息确认，这是通过AMQP协议来实现的。
通过将channel设置为confirm模式来实现

AMQP事务

在编程中与AMQP事务相关的主要有三个方法：

1
2
3

txSelect(); //将channel设置为transaction模式
txCommit(); //提交事务
txRollback(); //回滚事务

它的代码实现如下：

//消息生产者的实现
public class Send {

    private static final String QUEUE_NAME="queue_transition";

    public static void main(String[] args) throws IOException, TimeoutException {
        final Connection connection = ConnectionUtils.getConnection();
        final Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        //将当前channel设置为transition
        channel.txSelect();
        try {
            channel.basicPublish("",QUEUE_NAME,null,"消息内容".getBytes());
            //提交事务
            channel.txCommit();
        }catch (Exception e){
            System.out.println("事务归滚");
            //回滚事务
            channel.txRollback();
        }
        channel.close();
        connection.close();
    }
}

消息消费者的实现和普通的消息消费者没有区别。

消息发送的过程大致如下：

client发送tx.select
broker发送tx.select-ok
client发送消息到broker
client发送tx.commit
broker发送tx.commit-ok

如果在发送消息的时候发生了一次，进行了回滚，那么整个流程如下：

client发送tx.select
broker发送tx.select-ok
client发送消息到broker
client发送tx.rollback
broker发送tx.rollback-ok

因为在消息发送到队列之前就回滚了，所以接收端也不会收到消息了。当我们发现事务回滚了的时候，就可以进行相应的补救措施。

Confirm模式

使用AMQP事务来实现消息的确认会极大的损失性能。我们可以使用Confirm模式这种性能更高的做法。
Confirm可以分为两个方面，一方面是消息的生产者的confirm模式，一方面是消息消费者的confirm模式。消息消费者的confirm模式即手动消息确认和自动消息确认。

消息消费者的消息确认：

// 手动确认消息
channel.basicAck(envelope.getDeliveryTag(), false);

// 关闭自动确认
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck, consumer);

消息发送的原理图如下；

channel要么设置为confirm模式，要么设置为事务模式。
在消息生产者的信道上设置为confirm模式，一但信道进入confirm模式，那么在该信道上发送的所有的消息都会被指派一个唯一的ID，一但消息被投递到所有匹配的队列之后，Broker就会发送一个确认给消息生产者。
Confirm模式的一大好处在于，它是异步的，发送了消息之后可以在等待确认的同时继续发送另一条消息。

普通Confirm模式

消息发送者的实现：

public class Send {

    private static final String QUEUE_NAME="confirm_queue";

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {

        final Connection connection = ConnectionUtils.getConnection();

        final Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);

        //将channel设置为confirm模式
        channel.confirmSelect();

        for(int i=0;i<10;i++){

            channel.basicPublish("",QUEUE_NAME, MessageProperties.PERSISTENT_BASIC,("第"+i+"条消息").getBytes());

            if(channel.waitForConfirms()){
                System.out.println("消息发送成功");
            }else {
                //消息发送失败的处理
            }
        }
        channel.close();
        connection.close();


    }
}

消息接收者的实现：


public class Recv {
    private static  final String QUEUE_NAME="confirm_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        final Connection connection = ConnectionUtils.getConnection();
        final Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);

        final DefaultConsumer consumer = new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println(new String(body, "utf-8"));

            }
        };

        channel.basicConsume(QUEUE_NAME,true,consumer);
    }
}

批量Confirm模式

消息生产者的不同之处：

channel.confirmSelect();

for(int i=0;i<10;i++){

    channel.basicPublish("",QUEUE_NAME, MessageProperties.PERSISTENT_BASIC,("第"+i+"条消息").getBytes());
    
}
//发送一批消息之后，才进行确认
if(channel.waitForConfirms()){
    System.out.println("消息发送成功");
}else {
    //消息发送失败的处理
}

我们也可以在消息发送者中对每条消息进行监听处理：

channel.addConfirmListener(new ConfirmListener() {
    public void handleAck(long l, boolean b) throws IOException {
        System.out.println("消息"+l+"投递成功");
    }

    public void handleNack(long l, boolean b) throws IOException {
        System.out.println("消息"+l+"投递失败");

    }
});

异步Confirm模式

public class Send {

    private static final String QUEUE_NAME="confirm_queue";
    private  static SortedSet<Long> confirmSet= Collections.synchronizedSortedSet(new TreeSet<Long>());

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {

        final Connection connection = ConnectionUtils.getConnection();

        final Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        
        //将channel设置为confirm模式
        channel.confirmSelect();
        channel.addConfirmListener(new ConfirmListener() {
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                if (multiple) {
                    confirmSet.headSet(deliveryTag + 1L).clear();
                } else {
                    confirmSet.remove(deliveryTag);
                }

            }

            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                if (multiple) {
                    confirmSet.headSet(deliveryTag + 1L).clear();
                } else {
                    confirmSet.remove(deliveryTag);
                }
            }
        });

        for(int i=0;i<10;i++){
            final long no = channel.getNextPublishSeqNo();

            channel.basicPublish("",QUEUE_NAME, MessageProperties.PERSISTENT_BASIC,("第"+i+"条消息").getBytes());

            confirmSet.add(no);
        }
        channel.close();
        connection.close();


    }
}