4. 发布确认

4.1. 发布确认原理

生产者将信道设置成 confirm 模式，一旦信道进入 confirm 模式，**所有在该信道上面发布的消息都将会被指派一个唯一的 ID( 从 1 开始)，**一旦消息被投递到所有匹配的队列之后，broker 就会发送一个确认给生产者(包含消息的唯一 ID)，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker 回传给生产者的确认消息中 delivery-tag 域包含了确认消息的序列号，此外 broker 也可以设置basic.ack 的 multiple 域，表示到这个序列号之前的所有消息都已经得到了处理。

confirm 模式最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果 RabbitMQ 因为自身内部错误导致消息丢失，就会发送一条 nack 消息，生产者应用程序同样可以在回调方法中处理该 nack 消息。

4.2. 发布确认的策略

4.2.1. 开启发布确认的方法

发布确认默认是没有开启的，如果要开启需要调用方法 confirmSelect，每当你要想使用发布确认，都需要在 channel 上调用该方法

4.2.2. 单个确认发布

这是一种简单的确认方式，它是一种同步确认发布的方式，也就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布,waitForConfirmsOrDie(long)这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。

这种确认方式有一个最大的缺点就是:发布速度特别的慢，因为如果没有确认发布的消息就会阻塞所有后续消息的发布，这种方式最多提供每秒不超过数百条发布消息的吞吐量。当然对于某些应用程序来说这可能已经足够了。

public static void publishMessageIndividually() throws Exception { try (Channel channel = RabbitMqUtils.getChannel())
{ String queueName = UUID.randomUUID().toString(); channel.queueDeclare(queueName, false, false, false, null);
//开启发布确认 channel.confirmSelect();
long begin = System.currentTimeMillis(); for (int i = 0; i < MESSAGE_COUNT; i++)
{ String message = i + "";
channel.basicPublish("", queueName, null, message.getBytes());
//服务端返回 false 或超时时间内未返回，生产者可以消息重发
boolean flag = channel.waitForConfirms(); if(flag){
System.out.println("消息发送成功");
}
}
long end = System.currentTimeMillis();
System.out.println("发布" + MESSAGE_COUNT + "个单独确认消息,耗时" + (end - begin) +"ms");
}
}

package com.atguigu.rabbitmq.four;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.Channel;
import java.util.Date;
import java.util.UUID;
/**
 * 验证发布确认模式
 *
 * 1.单个确认
 *
 * 使用时间 比较那种确认方式最好 发布1000个消息单独确认消息，耗时5976ms
 *
 *
 * 2.批量确认
 *
 *  3.异步确认
 *
 */
public class ConfirmMessage {
    //批量发消息的个数
    public static final int MESSAGE_COUNT=1000;
    public static void main(String[] args) throws Exception {
      //  1.单个确认
        ConfirmMessage.publishMessageIndividuallu();
    //    2.批量确认
      //  3.异步确认
    }
//单个确认
    public static void publishMessageIndividuallu()throws  Exception{
        Channel channel = RabbitMqutils.getChannel();
        //队列的声明
        String quenuName = UUID.randomUUID().toString();
        System.out.println("quenuName-+++"+quenuName);
        channel.queueDeclare(quenuName,true,false,false,null);
        //开启确认发布的，模式
        channel.confirmSelect();
        //记录时间 开始的时间
        long l = System.currentTimeMillis();//毫秒的
        //批量的发消息
        //MESSAGE_COUNT.fori 就可以出现下面了省略的写法
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message=i+"";
            channel.basicPublish("",quenuName,null,message.getBytes());
            //单个消息确认
            boolean flag=channel.waitForConfirms();
            if (flag){
                System.out.println("消息发送成功");
            }
        }
        //结束时间 开始的时间
        long e = System.currentTimeMillis();//毫秒的
        System.out.println("发布"+MESSAGE_COUNT+"个消息单独确认消息，耗时"+(e-l)+"ms");
    }
}

4.2.3. 批量确认发布

上面那种方式非常慢，与单个等待确认消息相比，先发布一批消息然后一起确认可以极大地提高吞吐量，当然这种方式的缺点就是:当发生故障导致发布出现问题时，不知道是哪个消息出现问题了，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息。当然这种方案仍然是同步的，也一样阻塞消息的发布。

public static void publishMessageBatch() throws Exception { try (Channel channel = RabbitMqUtils.getChannel())
{ String queueName = UUID.randomUUID().toString(); channel.queueDeclare(queueName, false, false, false, null);
//开启发布确认 channel.confirmSelect();
//批量确认消息大小 int batchSize = 100;
//未确认消息个数 int outstandingMessageCount = 0;
long begin = System.currentTimeMillis(); for (int i = 0; i < MESSAGE_COUNT; i++)
{ String message = i + "";
channel.basicPublish("", queueName, null, message.getBytes()); outstandingMessageCount++;
if (outstandingMessageCount == batchSize) { channel.waitForConfirms(); outstandingMessageCount = 0;
}
}
//为了确保还有剩余没有确认消息 再次确认 if (outstandingMessageCount > 0)
{ channel.waitForConfirms();
}
long end = System.currentTimeMillis();
System.out.println("发布" + MESSAGE_COUNT + "个批量确认消息,耗时" + (end - begin) +
 "ms");
}
}
 package com.atguigu.rabbitmq.four;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.Channel;
import java.util.Date;
import java.util.UUID;
/**
 * 验证发布确认模式
 *
 * 1.单个确认
 *
 * 使用时间 比较那种确认方式最好 发布1000个消息单独确认消息，耗时5976ms
 *
 *
 * 2.批量确认
 *发布1000个消息批量确认消息，耗时362ms
 *  3.异步确认
 *
 */
public class ConfirmMessage {
    //批量发消息的个数
    public static final int MESSAGE_COUNT=1000;
    public static void main(String[] args) throws Exception {
      //  1.单个确认
      //  ConfirmMessage.publishMessageIndividuallu();
    //    2.批量确认
        ConfirmMessage.publishMessageBatch();
      //  3.异步确认
    }
//单个确认
    public static void publishMessageIndividuallu()throws  Exception{
        Channel channel = RabbitMqutils.getChannel();
        //队列的声明
        String quenuName = UUID.randomUUID().toString();
        System.out.println("quenuName-+++"+quenuName);
        channel.queueDeclare(quenuName,true,false,false,null);
        //开启确认发布的，模式
        channel.confirmSelect();
        //记录时间 开始的时间
        long l = System.currentTimeMillis();//毫秒的
        //批量的发消息
        //MESSAGE_COUNT.fori 就可以出现下面了省略的写法
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message=i+"";
            channel.basicPublish("",quenuName,null,message.getBytes());
            //单个消息确认
            boolean flag=channel.waitForConfirms();
            if (flag){
                System.out.println("消息发送成功");
            }
        }
        //结束时间 开始的时间
        long e = System.currentTimeMillis();//毫秒的
        System.out.println("发布"+MESSAGE_COUNT+"个消息单独确认消息，耗时"+(e-l)+"ms");
    }
    //批量发布确认
public static void publishMessageBatch()throws Exception{
    Channel channel = RabbitMqutils.getChannel();
    //队列的声明
    String quenuName = UUID.randomUUID().toString();
    System.out.println("quenuName-+++"+quenuName);
    channel.queueDeclare(quenuName,true,false,false,null);
    //开启确认发布的，模式
    channel.confirmSelect();
    //记录时间 开始的时间
    long l = System.currentTimeMillis();//毫秒的
    //批量确认的长度
    int batshSize=100;
    //未确认的个数
//批量发送消息 批量发布确认
    for (int i = 0; i < MESSAGE_COUNT; i++) {
        String message=i+"";
        channel.basicPublish("",quenuName,null,message.getBytes());
      //判断达到消息100条消息的时候 批量确认一次
        if(i%batshSize==0){
            //消息确认
            channel.waitForConfirms();
        }
    }
    //结束时间 开始的时间
    long e = System.currentTimeMillis();//毫秒的
    System.out.println("发布"+MESSAGE_COUNT+"个消息批量确认消息，耗时"+(e-l)+"ms");
}
}

4.2.4. 异步确认发布

**异步确认虽然编程逻辑比上两个要复杂，但是性价比最高，无论是可靠性还是效率都没得说，他是利用回调函数来达到消息可靠性传递的，这个中间件也是通过函数回调来保证是否投递成功，**下面就让我们来详细讲解异步确认是怎么实现的。

public static void publishMessageAsync() throws Exception {
 try (Channel channel = RabbitMqUtils.getChannel())
{ String queueName = UUID.randomUUID().toString(); 
channel.queueDeclare(queueName, false, false, false, null);
//开启发布确认 channel.confirmSelect();
/**
* 线程安全有序的一个哈希表，适用于高并发的情况
*  1.轻松的将序号与消息进行关联
*  2.轻松批量删除条目 只要给到序列号
*  3.支持并发访问
*/
ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
/**
* 确认收到消息的一个回调
*  1.消息序列号
*  2.true 可以确认小于等于当前序列号的消息
*    false 确认当前序列号消息
*/
ConfirmCallback ackCallback = (sequenceNumber, multiple) -> 
{ if (multiple) {
//返回的是小于等于当前序列号的未确认消息 是一个 map
ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(sequenceNumber, true);
//清除该部分未确认消息 confirmed.clear();
}else{
//只清除当前序列号的消息
outstandingConfirms.remove(sequenceNumber);
}
};
ConfirmCallback nackCallback = (sequenceNumber, multiple) -> { String message = outstandingConfirms.get(sequenceNumber);
System.out.println("发布的消息"+message+"未被确认，序列号"+sequenceNumber);
};
/**
* 添加一个异步确认的监听器
*  1.确认收到消息的回调
*  2.未收到消息的回调
*/
channel.addConfirmListener(ackCallback, null);
long begin = System.currentTimeMillis(); for (int i = 0; i < MESSAGE_COUNT; i++)
{ String message = "消息" + i;
/**
*   channel.getNextPublishSeqNo()获取下一个消息的序列号
* 通过序列号与消息体进行一个关联
* 全部都是未确认的消息体
*/
outstandingConfirms.put(channel.getNextPublishSeqNo(), message); channel.basicPublish("", queueName, null, message.getBytes());
}
long end = System.currentTimeMillis();
System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息,耗时" + (end - begin) +
"ms");
}
}
package com.atguigu.rabbitmq.four;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import java.util.Date;
import java.util.UUID;
/**
 * 验证发布确认模式
 *
 * 1.单个确认
 *
 * 使用时间 比较那种确认方式最好 发布1000个消息单独确认消息，耗时5976ms
 *
 *
 * 2.批量确认
 *发布1000个消息批量确认消息，耗时362ms
 *  3.异步确认
 *发布1000个消息异步发布确认消息，耗时73ms
 */
public class ConfirmMessage {
    //批量发消息的个数
    public static final int MESSAGE_COUNT=1000;
    public static void main(String[] args) throws Exception {
      //  1.单个确认
      //  ConfirmMessage.publishMessageIndividuallu();
    //    2.批量确认
     //   ConfirmMessage.publishMessageBatch();
      //  3.异步确认
        ConfirmMessage.publishMEssageAsync();
    }
//单个确认
    public static void publishMessageIndividuallu()throws  Exception{
        Channel channel = RabbitMqutils.getChannel();
        //队列的声明
        String quenuName = UUID.randomUUID().toString(); //javaJDK提供的一个自动生成主键的方法
        System.out.println("quenuName-+++"+quenuName);
        channel.queueDeclare(quenuName,true,false,false,null);
        //开启确认发布的，模式
        channel.confirmSelect();
        //记录时间 开始的时间
        long l = System.currentTimeMillis();//毫秒的
        //批量的发消息
        //MESSAGE_COUNT.fori 就可以出现下面了省略的写法
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message=i+"";
            channel.basicPublish("",quenuName,null,message.getBytes());
            //单个消息确认
            boolean flag=channel.waitForConfirms();
            if (flag){
                System.out.println("消息发送成功");
            }
        }
        //结束时间 开始的时间
        long e = System.currentTimeMillis();//毫秒的
        System.out.println("发布"+MESSAGE_COUNT+"个消息单独确认消息，耗时"+(e-l)+"ms");
    }
    //批量发布确认
public static void publishMessageBatch()throws Exception{
    Channel channel = RabbitMqutils.getChannel();
    //队列的声明
    String quenuName = UUID.randomUUID().toString();
    System.out.println("quenuName-+++"+quenuName);
    channel.queueDeclare(quenuName,true,false,false,null);
    //开启确认发布的，模式
    channel.confirmSelect();
    //记录时间 开始的时间
    long l = System.currentTimeMillis();//毫秒的
    //批量确认的长度
    int batshSize=100;
    //未确认的个数
//批量发送消息 批量发布确认
    for (int i = 0; i < MESSAGE_COUNT; i++) {
        String message=i+"";
        channel.basicPublish("",quenuName,null,message.getBytes());
      //判断达到消息100条消息的时候 批量确认一次
        if(i%batshSize==0){
            //消息确认
            channel.waitForConfirms();
        }
    }
    //结束时间 开始的时间
    long e = System.currentTimeMillis();//毫秒的
    System.out.println("发布"+MESSAGE_COUNT+"个消息批量确认消息，耗时"+(e-l)+"ms");
}
//异步发布确认
    public static void publishMEssageAsync()throws Exception{
        Channel channel = RabbitMqutils.getChannel();
        //队列的声明
        String quenuName = UUID.randomUUID().toString(); //javaJDK提供的一个自动生成主键的方法
        System.out.println("quenuName-+++"+quenuName);
        channel.queueDeclare(quenuName,true,false,false,null);
        //开启确认发布的，模式
        channel.confirmSelect();
        //记录时间 开始的时间
        long l = System.currentTimeMillis();//毫秒的
         //消息确认成功 回调 函数
        ConfirmCallback ackCallback= (var1,var3)->{
            System.out.println("确认的消息"+var1);
        };
        //消息确认失败 回调 函数
        /**
         * 1.消息的标记  var1
         * 、2.是否批量删除  var3
         */
        ConfirmCallback nackCallback=(var1,var3)->{
            System.out.println("未确认的消息"+var1);
        };
     //准备消息的监听器 监听纳西人消息成功了 那些消息失败了呢
        /**
         * 1.监听成功的 监听那些是成的消息
         * 2.监听失败的消息，监听那些消息是失败的
         */
      channel.addConfirmListener(ackCallback,nackCallback); //异步监听
    //批量发送消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message="消息"+i;
            channel.basicPublish("",quenuName,null,message.getBytes());
        }
        //结束时间 开始的时间
        long e = System.currentTimeMillis();//毫秒的
        System.out.println("发布"+MESSAGE_COUNT+"个消息异步发布确认消息，耗时"+(e-l)+"ms");
    }
}

全部代码

package com.atguigu.rabbitmq.four;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import java.util.Date;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
/**
 * 验证发布确认模式
 * <p>
 * 1.单个确认
 * <p>
 * 使用时间 比较那种确认方式最好 发布1000个消息单独确认消息，耗时5976ms
 * <p>
 * <p>
 * 2.批量确认
 * 发布1000个消息批量确认消息，耗时362ms
 * 3.异步确认
 * 发布1000个消息异步发布确认消息，耗时73ms
 * 发布1000个消息异步发布确认消息，耗时59ms
 */
public class ConfirmMessage {
    //批量发消息的个数
    public static final int MESSAGE_COUNT = 1000;
    public static void main(String[] args) throws Exception {
        //  1.单个确认
        //  ConfirmMessage.publishMessageIndividuallu();
        //    2.批量确认
        //   ConfirmMessage.publishMessageBatch();
        //  3.异步确认
        ConfirmMessage.publishMEssageAsync();
    }
    //单个确认
    public static void publishMessageIndividuallu() throws Exception {
        Channel channel = RabbitMqutils.getChannel();
        //队列的声明
        String quenuName = UUID.randomUUID().toString(); //javaJDK提供的一个自动生成主键的方法
        System.out.println("quenuName-+++" + quenuName);
        channel.queueDeclare(quenuName, true, false, false, null);
        //开启确认发布的，模式
        channel.confirmSelect();
        //记录时间 开始的时间
        long l = System.currentTimeMillis();//毫秒的
        //批量的发消息
        //MESSAGE_COUNT.fori 就可以出现下面了省略的写法
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", quenuName, null, message.getBytes());
            //单个消息确认
            boolean flag = channel.waitForConfirms();
            if (flag) {
                System.out.println("消息发送成功");
            }
        }
        //结束时间 开始的时间
        long e = System.currentTimeMillis();//毫秒的
        System.out.println("发布" + MESSAGE_COUNT + "个消息单独确认消息，耗时" + (e - l) + "ms");
    }
    //批量发布确认
    public static void publishMessageBatch() throws Exception {
        Channel channel = RabbitMqutils.getChannel();
        //队列的声明
        String quenuName = UUID.randomUUID().toString();
        System.out.println("quenuName-+++" + quenuName);
        channel.queueDeclare(quenuName, true, false, false, null);
        //开启确认发布的，模式
        channel.confirmSelect();
        //记录时间 开始的时间
        long l = System.currentTimeMillis();//毫秒的
        //批量确认的长度
        int batshSize = 100;
        //未确认的个数
//批量发送消息 批量发布确认
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", quenuName, null, message.getBytes());
            //判断达到消息100条消息的时候 批量确认一次
            if (i % batshSize == 0) {
                //消息确认
                channel.waitForConfirms();
            }
        }
        //结束时间 开始的时间
        long e = System.currentTimeMillis();//毫秒的
        System.out.println("发布" + MESSAGE_COUNT + "个消息批量确认消息，耗时" + (e - l) + "ms");
    }
    //异步发布确认
    public static void publishMEssageAsync() throws Exception {
        Channel channel = RabbitMqutils.getChannel();
        //队列的声明
        String quenuName = UUID.randomUUID().toString(); //javaJDK提供的一个自动生成主键的方法
        System.out.println("quenuName-+++" + quenuName);
        channel.queueDeclare(quenuName, true, false, false, null);
        //开启确认发布的模式
        channel.confirmSelect();
        /**
         * 线程安全由于的一个 哈希表 适用于到并发的情况下
         * 选择他的理由
         * 1.他可以轻松的 将序号和消息关联 long ,String
         * 2.可以轻松的批量的删除条目  只要给我key就可以删除value  ()
         *3.支持高并发 （多线程）
         */
        ConcurrentSkipListMap<Long,String> outstandingConfirms=
                new ConcurrentSkipListMap<>();
        //消息确认成功 回调 函数
        ConfirmCallback ackCallback = (var1, var3) -> {
            //2.删除掉确认的消息 剩下就是未确认的消息
    if(var3){
        //是批量的话 调用id删除
        ConcurrentNavigableMap<Long, String> confirmed =
                outstandingConfirms.headMap(var1);
        confirmed.clear();
    }else{
  //不是的话 直接删除
        outstandingConfirms.remove(var1);
    }
            System.out.println("确认的消息" + var1);
        };
        //消息确认失败 回调 函数
        /**
         * 1.消息的标记  var1
         * 、2.是否批量删除  var3  true/false
         */
        ConfirmCallback nackCallback = (var1, var3) -> {
            //3.打印未确认的消息有哪些
            String messgae = outstandingConfirms.get(var1);
            System.out.println("未确认的消息是"+messgae+"----未确认的消息tag" + var1);
        };
        //准备消息的监听器 监听纳西人消息成功了 那些消息失败了呢
        /**
         * 1.监听成功的 监听那些是成的消息
         * 2.监听失败的消息，监听那些消息是失败的
         */
        channel.addConfirmListener(ackCallback, nackCallback); //异步监听
        //记录时间 开始的时间
        long l = System.currentTimeMillis();//毫秒的
        //批量发送消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            channel.basicPublish("", quenuName, null, message.getBytes());
            //1.此处要记录下所有要发送消息的 总和
            // 序号  信息
            outstandingConfirms.put(channel.getNextPublishSeqNo(),message);
        }
        //结束时间 开始的时间
        long e = System.currentTimeMillis();//毫秒的
        System.out.println("发布" + MESSAGE_COUNT + "个消息异步发布确认消息，耗时" + (e - l) + "ms");
    }
}

4.2.5. 如何处理异步未确认消息

最好的解决的解决方案就是把未确认的消息放到一个基于内存的能被发布线程访问的队列，比如说用 ConcurrentLinkedQueue 这个队列在 confirm callbacks 与发布线程之间进行消息的传递。

4.2.6. 以上 3 种发布确认速度对比

单独发布消息

同步等待确认，简单，但吞吐量非常有限。

批量发布消息

批量同步等待确认，简单，合理的吞吐量，一旦出现问题但很难推断出是那条

消息出现了问题。

异步处理：

最佳性能和资源使用，在出现错误的情况下可以很好地控制，但是实现起来稍微难些

public static void main(String[] args) throws Exception {
//这个消息数量设置为 1000 好些 不然花费时间太长 publishMessagesIndividually(); publishMessagesInBatch(); handlePublishConfirmsAsynchronously();
}
//运行结果
发布 1,000 个单独确认消息耗时 50,278 ms
发布 1,000 个批量确认消息耗时 635 ms
发布 1,000 个异步确认消息耗时 92 ms

RibbitMQ学习笔记之MQ发布确认

4. 发布确认

4.1. 发布确认原理

4.2. 发布确认的策略

4.2.1. 开启发布确认的方法

4.2.2. 单个确认发布

4.2.3. 批量确认发布

4.2.4. 异步确认发布

4.2.5. 如何处理异步未确认消息

4.2.6. 以上 3 种发布确认速度对比

热门文章

最新文章

相关课程

相关电子书

相关实验场景