系统学习消息队列——RabbitMQ的消息发布确认

1.MQ发送信息的时候产生的问题

2.MQ的发布确认原理

3.MQ的发布确认策略

1.MQ发送信息的时候产生的问题

我们在前一篇博客系统学习消息队列——RabbitMQ的消息应答和持久化中学过，当消费者挂掉的时候，有消息重发，当队列挂掉的时候，有消息持久化，但是我们却无法保证生产者发送到队列的消息能否确定发送成功，这个时候就有了消息的发布确认。

2.MQ的发布确认原理
当我们的信道被设置成发布确认（confirm）模式，那么所有在该信道上面发布的消息都会被指派一个唯一的ID，一旦消息成功投递，broker就会发送一个确认给生产者，生产者此时就知道消息已经投递成功，生产者就会把这条消息进行删除。

confirm模式可以是同步的，也可以是异步的，同步的情况下是发送之后马上进行确认，异步的话生产者可以无需等待确认只管发送消息，如果某些消息得到确认，生产者将就可以通过回调方法来确认消息。

3.MQ的发布确认策略

3.1）开启确认发布
发布确认模式默认是没有开启的，我们需要调用方法将它打开。

        Channel channel = connection.createChannel();
        开启发布确认
        channel.confirmSelect();

3.2）单个确认发布

这是一种简单的同步确认方式，发送一条消息，确认一条消息，后续的消息才能继续发送。

优点：简单易懂。
缺点：发布速度过慢，如果前面的消息没有得到确认，后面的消息就不得发送，容易阻塞。

public class ProducerSingle {
private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {

//channel 实现了自动 close 接口 自动关闭 不需要自己进行关闭
try (Channel channel = RabbitMqUtils.getChannel()) {
/**
             * 声明一个队列
             * 1.队列名称
             * 2.队列里面的消息是否持久化 默认消息存储在内存中
             * 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费
             * 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除
             * 5.其他参数
             */
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
//开始时间
            long begin = System.currentTimeMillis();
for (int i = 0; i < 1000; i++) {
                String message = i + "";
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
//确认是否发送成功，服务端返回 false 或超时时间内未返回，生产者可以消息重发
                boolean flag = channel.waitForConfirms();
if (flag) {
                    System.out.println("消息发送成功");
                }
            }
//发送结束时间
            long end = System.currentTimeMillis();
            System.out.println("发布" + 1000 + "个单独确认消息,耗时" + (end - begin) + "ms");
        }
    }
}

3.3）批量确认发布
单个确认发布的速度非常慢，其实我们可以先发送一批，然后确认一批，再发布一批。
优点：比单个确认发布速度快，吞吐量大。
缺点：当其中一个消息出问题的时候，不知道是哪个消息出现了问题，我们必须将整个批处理消息保存在内存里，以记录重要的消息后重新发布消息。这种方法也是阻塞的，一样阻塞消息的发布。

public class ProducerMulti {
private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {

//channel 实现了自动 close 接口 自动关闭 不需要自己进行关闭
try (Channel channel = RabbitMqUtils.getChannel()) {
/**
             * 声明一个队列
             * 1.队列名称
             * 2.队列里面的消息是否持久化 默认消息存储在内存中
             * 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费
             * 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除
             * 5.其他参数
             */
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);

//批量确认消息大小
int batchSize = 100;
//未确认消息个数
int unConfirmMessageNum = 0;
//开始发送时间
            long begin = System.currentTimeMillis();
for (int i = 0; i < 1000; i++) {
                String message = i + "";
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
//发送一条消息，未确认书+1
                unConfirmMessageNum++;
//如果位确认数到达批量确认大小
if (unConfirmMessageNum == batchSize) {
//等待进行批量确认
                    channel.waitForConfirms();
                    unConfirmMessageNum = 0;
                }
            }
//为了确保还有剩余没有确认消息 再次确认
if (unConfirmMessageNum > 0) {
                channel.waitForConfirms();
            }
//结束时间
            long end = System.currentTimeMillis();
            System.out.println("发布" + 1000 + "个单独确认消息,耗时" + (end - begin) + "ms");
        }
    }
}

3.4）异步确认发布
异步确认不需要阻塞，生产者只管发送信息就好，队列通过回调函数通知生产者发送成功。

优点：保证了效率和可靠性

缺点：编程逻辑复杂

public class ProducerAsyn {
private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {

//channel 实现了自动 close 接口 自动关闭 不需要自己进行关闭
try (Channel channel = RabbitMqUtils.getChannel()) {
/**
             * 声明一个队列
             * 1.队列名称
             * 2.队列里面的消息是否持久化 默认消息存储在内存中
             * 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费
             * 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除
             * 5.其他参数
             */
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);

/**
             * 用于回调函数确认发布的哈希表，线程安全，适用于并发情况
             * 1.可以将序列号和消息进行关联
             * 2.可以批量删除已经确认的消息
             * 3.支持并发访问
             */
            ConcurrentSkipListMap<Long, String> confirmsMap = new ConcurrentSkipListMap<>();

/**
             * 确认收到消息的一个回调
             * 1.消息序列号
             * 2.true 可以确认小于等于当前序列号的消息
             *   false 确认当前序列号消息
             */
            ConfirmCallback ackCallback = (sequenceNumber, multiple) -> {
if (multiple) {
//把小于当前序列号的全部消息取出
//返回的是小于等于当前序列号的未确认消息 是一个 map
                    ConcurrentNavigableMap<Long, String> confirmed =
                            confirmsMap.headMap(sequenceNumber, true);
//清除该部分未确认消息
                    confirmed.clear();
                }else{
//只清除当前序列号的消息
                    confirmsMap.remove(sequenceNumber);
                }
            };
//未被确认的回调
            ConfirmCallback nackCallback = (sequenceNumber, multiple) ->
            {String message = confirmsMap.get(sequenceNumber);
                System.out.println("发布的消息"+message+"未被确认，序列号"+sequenceNumber);
            };
/**
             * 添加一个异步确认的监听器
             * 1.确认收到消息的回调
             * 2.未收到消息的回调
             */
            channel.addConfirmListener(ackCallback, nackCallback);

//发送开始时间
            long begin = System.currentTimeMillis();
for (int i = 0; i < 1000; i++) {
                String message = i + "";
//在map里面设置消息id和内容
                confirmsMap.put(channel.getNextPublishSeqNo(), message);
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            }
//发送结束时间
            long end = System.currentTimeMillis();
            System.out.println("发布" + 1000 + "个单独确认消息,耗时" + (end - begin) + "ms");
        }
    }
}

3.5）确认发布速度对比

版权声明：本文内容来自个人博客segmentfault:苏凌峰，遵循CC 4.0 BY-SA版权协议上原文接及本声明。
本作品采用知识共享署名-非商业性使用-禁止演绎 2.5 中国大陆许可协议进行可。
原文链接：https://segmentfault.com/a/1190000041987998
如有涉及到侵权，请联系，将立即予以删除处理。
在此特别鸣谢原作者的创作。
此篇文章的所有版权归原作者所有，与本公众号无关，商业转载建议请联系原作者，非商业转载请注明出处。