RabbitMQ高级特性之消息可靠性

野性酷女 2022-08-31 12:30 209阅读 0赞

## 消息可靠性 ##

如何确保消息的可靠性？  
支付场景中，我们在超时后买商品之后，扫码枪将二维码扫描之后，会给我们推送一条消息，说购买成功，但此时银行还未进行扣款，过来一段时间之后，银行会给我们发来短信通知消费多少钱。

这种场景中，我们如何保证我们的消息不被丢失？

可以从以下几方面来保证消息的可靠性：

1.  客户端代码中的异常捕获，包括生产者和消费者
2.  AMQP/RabbitMQ的事务机制
3.  发送端确认机制
4.  消息持久化机制
5.  Broker端的高可用集群
6.  消费者确认机制
7.  消费端限流
8.  消息幂等性

> 还可以使用RabbitMq事务机制来保整消息可靠性，但不推荐，原因是开销太大，会导致mq性能下降。

流程图

生产者就是我们的扫码枪，扫码之后将我们的支付信息丢到Mq中，然后进行后续的操作。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70]  
1 Send到mq中时候，消息不一定成功，如果发生异常，我们可以重新扫码。  
2 发送到mq中，这一过程，我们发送了，但不能确保100%发送成功，只是将消息发出去。此时需要mq给我返回一个ack回执信息，我们才能确保消息100%是发送成功了。  
3 发送到mq中就一定是消息不丢失吗？不一定，如果mq没有做持久化操作呢？如果队列，交换机 没有持久化呢？ 如果是单机mq呢？这种情况下我们的消息还是会丢失的。  
4 消费阶段，同理消费成功了，返回ack给mq ，mq将其删除。

针对mq这中情况，可以做到数据最终一致性，最终我们的钱还是会扣掉的。  
优点，异步操作，支持高并发操作，数据最终一致性。  
缺点，消息多的情况下，可能会存在延迟，消息堆积。

### 如何实现呢？ ###

1 生产者发送消息 异常捕获机制

try{ 
    //发送消息
    //判断
    if(true){ 
    }
    //没异常不能保证100%绝对可靠
    }catch(){ 
    //重试代码
    }

`可以在捕获后手动去重试`  
`可以通过spring.rabbitmq.template.retry.enabled=true 配置开启发送端的重试`

2 发送端确认机制 mq返回确认

发送消息后，由mq返回确认ack给发送端

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 1]

RabbitMQ后来引入了一种轻量量级的方式，叫发送方确认(publisher confirm)机制。  
生产者将信道设置成confirm(确认)模式，一旦信道进入confirm 模式，所有在该信道上⾯面发布的消息都会被指派一个唯一的ID(从1 开始)，一旦消息被投递到所有匹配的队列之后（如果消息和队列是持久化的，那么确认消息会在消息持久化后发出），RabbitMQ 就会发送一个确认(Basic.Ack)给生产者(包含消息的唯一ID)，这样生产者就知道消息已经正确送达了。

生产者同步等待，获取消息ack，如没异常就表示消息确认。  
\`等待确认过程中会处于阻塞状态\`\`  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 2]

RabbitMQ 回传给生产者的确认消息中的deliveryTag 字段包含了确认消息的序号，另外，通过设置channel.basicAck方法中的multiple参数，表示到这个序号之前的所有消息是否都已经得到了处理  
了。

### 具体操作 ###

实现代码

#### 同步等待确认 ####

可以通过“批处理理”的方式来改善整体的性能（即批量量发送消息后仅调用一次waitForConfirms方法）。正常情况下这种批量处理的方式效率会高很多，但是如果发生了超时或者nack（失败）后那就需要批量量重发消息或者通知上游业务批量回滚（因为我们只知道这个批次中有消息没投递成功，而并不知道具体是那条消息投递失败了，所以很难针对性处理）。

`其他监听器`

当mq路由到不存在的队列时，Mq会直接将消息丢掉，我们可以通过监听来讲消息打印出来

前提在发送消息时设置mandatory标志，即可开启故障检测模式，这个不能保证发送消息ack100%成功。

![在这里插入图片描述][44212c4148884b9cabb919bf4b54cd59.png]

打印

channel.addReturnListener(new ReturnListener() { 
                @Override
                public void handleReturn(int replyCode, String replyText, String exchange, String routingKey,
                                         AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("replyText: " +replyText);
                    System.out.println("exchange: " + exchange);
                    System.out.println("routingKey: " + routingKey);
                    System.out.println("message: " + new String(body));
                }
            });

信道关闭时触发

channel.addShutdownListener(new ShutdownListener() { 
        @Override
        public void shutdownCompleted(ShutdownSignalException cause) { 
            System.out.println(cause.getMessage());
        }
    });

连接关闭时触发

connection.addShutdownListener(new ShutdownListener() { 
        @Override
        public void shutdownCompleted(ShutdownSignalException cause) { 
            System.out.println(cause.getMessage());
        }
    });

#### 批量同步等待确认 ####

批量重发消息肯定会造成部分消息重复。mq可以通过异步回调的方式来处理Broker的响应。addConfirmListener 方法可以添加ConfirmListener 这个回调接口，这个 ConfirmListener 接口包含两个方法:handleAck 和handleNack，分别用来处理 RabbitMQ 回传的 Basic.Ack 和 Basic.Nack。

public class ConfirmSelect02 { 
    
    
        public static void main(String[] args) throws NoSuchAlgorithmException, KeyManagementException, URISyntaxException, IOException, TimeoutException, InterruptedException { 
    
    
            //连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setUri("amqp://guest:guest@localhost:5672/%2f");
    
            //简历连接
            Connection connection = connectionFactory.newConnection();
    
            //获取通道
            Channel channel = connection.createChannel();
    
            //开启 确认信道
            channel.confirmSelect();
    
            //声明队列
            channel.queueDeclare("ex.tc.queue", true, false, false, null);
    
            //声明交换器
            channel.exchangeDeclare("ex.tc", BuiltinExchangeType.DIRECT, true, false, false, null);
    
    
            //绑定队列到交换器
            channel.queueBind("ex.tc.queue", "ex.tc", "key.queue");
    
    
            //批处理 确认消息
            String message = "hello-";
            // 批处理的大小
            int batchSize = 10;
            // 用于对需要等待确认消息的计数
            int outstrandingConfirms = 0;
            for (int i = 0; i < 103; i++) { 
                channel.basicPublish("ex.tc", "key.queue", null, (message + i).getBytes());
    
                outstrandingConfirms++;
                if (outstrandingConfirms == batchSize) { 
                    // 此时已经有一个批次的消息需要同步等待broker的确认消息
                    // 同步等待
                    channel.waitForConfirmsOrDie(5_000);
                    System.out.println("消息已经被确认了");
                    outstrandingConfirms = 0;
                }
            }
    
            if (outstrandingConfirms > 0) { 
                channel.waitForConfirmsOrDie(5_000);
                System.out.println("剩余消息已经被确认了");
            }
    
            channel.close();
            connection.close();
    
    
        }
    }

#### 异步回调等待确认 ####

package gaojitexing.kekaoxing;
    
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.net.URISyntaxException;
    import java.security.KeyManagementException;
    import java.security.NoSuchAlgorithmException;
    import java.util.concurrent.ConcurrentNavigableMap;
    import java.util.concurrent.ConcurrentSkipListMap;
    import java.util.concurrent.TimeoutException;
    
    /** * 批处理，异步回调方式确认消息 */
    public class ConfirmSelect03 { 
    
    
        public static void main(String[] args) throws NoSuchAlgorithmException, KeyManagementException, URISyntaxException, IOException, TimeoutException, InterruptedException { 
    
    
            //连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setUri("amqp://guest:guest@localhost:5672/%2f");
    
            //简历连接
            Connection connection = connectionFactory.newConnection();
    
            //获取通道
            Channel channel = connection.createChannel();
    
            //开启 确认信道
            channel.confirmSelect();
    
            //声明队列
            channel.queueDeclare("ex.tc.queue", true, false, false, null);
    
            //声明交换器
            channel.exchangeDeclare("ex.tc", BuiltinExchangeType.DIRECT, true, false, false, null);
    
    
            //绑定队列到交换器
            channel.queueBind("ex.tc.queue", "ex.tc", "key.queue");
    
    
            ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
    
    // ConfirmCallback clearOutstandingConfirms = new ConfirmCallback() { 
    // @Override
    // public void handle(long deliveryTag, boolean multiple) throws IOException { 
    // if (multiple) { 
    // System.out.println("编号小于等于 " + deliveryTag + " 的消息都已经被确认了");
    // } else { 
    // System.out.println("编号为：" + deliveryTag + " 的消息被确认");
    // }
    // }
    // };
    
            //异步回调 处理
            ConfirmCallback clearOutstandingConfirms = (deliveryTag, multiple) -> { 
                if (multiple) { 
                    System.out.println("编号小于等于 " + deliveryTag + " 的消息都已经被确认了");
                    final ConcurrentNavigableMap<Long, String> headMap
                            = outstandingConfirms.headMap(deliveryTag, true);
                    // 清空outstandingConfirms中已经被确认的消息信息
                    headMap.clear();
    
                } else { 
                    // 移除已经被确认的消息
                    outstandingConfirms.remove(deliveryTag);
                    System.out.println("编号为：" + deliveryTag + " 的消息被确认");
                }
            };
    
            // 设置channel的监听器，处理确认的消息和不确认的消息
            channel.addConfirmListener(clearOutstandingConfirms, (deliveryTag, multiple) -> { 
                if (multiple) { 
                    // 将没有确认的消息记录到一个集合中
                    // 此处省略实现
                    System.out.println("消息编号小于等于：" + deliveryTag + " 的消息 不确认");
                } else { 
                    System.out.println("编号为：" + deliveryTag + " 的消息不确认");
                }
            });
    
            String message = "hello-";
            for (int i = 0; i < 1000; i++) { 
                // 获取下一条即将发送的消息的消息ID
                final long nextPublishSeqNo = channel.getNextPublishSeqNo();
                channel.basicPublish("ex.tc", "key.queue", null, (message + i).getBytes());
                System.out.println("编号为：" + nextPublishSeqNo + " 的消息已经发送成功，尚未确认");
                outstandingConfirms.put(nextPublishSeqNo, (message + i));
            }
    
            // 等待消息被确认
            Thread.sleep(10000);
    
            channel.close();
            connection.close();
    
    
        }
    }

## 持久化存储机制 ##

持久化是提高RabbitMQ可靠性的基础，否则当RabbitMQ遇到异常时（如：重启、断电、停机等）数据将会丢失。主要从以下几个方面来保障消息的持久性：

1.  Exchange的持久化。通过定义时设置durable 参数为ture来保证Exchange相关的元数据不不丢失。
2.  Queue的持久化。也是通过定义时设置durable 参数为ture来保证Queue相关的元数据不不  
    丢失。
3.  消息的持久化。通过将消息的投递模式 (BasicProperties 中的 deliveryMode 属性)设置为 2  
    即可实现消息的持久化，保证消息自身不丢失。

队列和交换机的持久化可以通过代码来设置，true表示持久化，重启mq的时候不会删除队列和交换机。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 3]  
消息的持久化可以通过代码设置

AMQP.BasicProperties.Builder basicProperties =new  AMQP.BasicProperties.Builder();
                
                basicProperties.contentType("text/plain");//设置消息类型
                basicProperties.deliveryMode(2);//2表示持久化消息
                
                Map<String, Object> head = new HashMap<>();
                head.put("key","123");//可以设置属性值
                basicProperties.headers(head);
                
                AMQP.BasicProperties build = basicProperties.build();
                
                channel.basicPublish("ex.tc", "key.queue", build, (message + i).getBytes());

RabbitMQ中的持久化消息都需要写入磁盘（当系统内存不不足时，非持久化的消息也会被刷盘处理理），这些处理理动作都是在“持久层”中完成的。持久层是一个逻辑上的概念，实际包含两个部分：

1.  队列索引(rabbit\_queue\_index)，rabbit\_queue\_index 负责维护Queue中消息的信息，包括消息的存储位置、是否已交给消费者、是否已被消费及Ack确认等，每个Queue都有与之对应的rabbit\_queue\_index。
2.  消息存储(rabbit\_msg\_store)，rabbit\_msg\_store 以键值对的形式存储消息，它被所有队列列共享，在每个节点中有且只有一个。

那消息存储在哪儿呢？

RabbitMq home文件路径/msg\_stores/vhosts/$VHost/Id 这个路路径下包含 queues、msg\_store\_persistent、msg\_store\_transient 这 3 个目录，这是实际存储消息的位置。

其中`queues`目录中保存着`rabbit_queue_index`相关的数据，而`msg_store_persistent`保存着`持久化消息`数据，`msg_store_transient`保存着`非持久化`相关的数据。

如我本机目录：  
`/usr/local/var/lib/rabbitmq/mnesia/rabbit@localhost/msg_stores/vhosts/628WB79CIFDYO9LJI6DKMI09L`

628WB79CIFDYO9LJI6DKMI09L是`虚拟机id`目录，多个虚拟机的话会有多个。

我的文件  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 4]

由于我们的数据大小`小于4096byte`(默认) `直接落到索引文件`里，`没有刷到持久化文件`里，所以大小是`0字节`。  
`非持久化消息`的在`内存不足`的情况下才会刷到非持久化下文件里。

`注意`  
RabbitMQ通过配置`queue_index_embed_msgs_below`可以根据消息大小决定存储位置，  
默认`queue_index_embed_msgs_below`是`4096`字节(包含消息体、属性及headers)，小于该值的消息存在rabbit\_queue\_index中。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 5]

在索引文件里可以看到持久化的数据

代码示例

AMQP.BasicProperties.Builder basicProperties =new  AMQP.BasicProperties.Builder();
    
            basicProperties.contentType("text/plain");//设置消息类型
            basicProperties.deliveryMode(2);//2表示持久化消息
    
            Map<String, Object> head = new HashMap<>();
            head.put("key","123");//可以设置属性值
            basicProperties.headers(head);
    
            AMQP.BasicProperties build = basicProperties.build();
    
            channel.basicPublish("ex.tc", "key.queue", build, "hello world".getBytes());

我们存储的head以及消息

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 6]

## 消费端Consumer ACK ##

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70]: /images/20220829/f52cdf88bfcf4647ba64e327f91e13a6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 1]: /images/20220829/e7a968775cbe488c9e61361f9990aa97.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 2]: /images/20220829/ef503b6948514741a03b1a43c7dd616f.png
[44212c4148884b9cabb919bf4b54cd59.png]: /images/20220829/26df340329754aa693214ea03c431e32.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 3]: /images/20220829/336da73430f34904bf756a0617364f17.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 4]: /images/20220829/27d47272a74b457b8019e0f38a7d00d2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 5]: /images/20220829/da1ca3c0ac9e44a7be7e5ce82e9e59b9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 6]: /images/20220829/2324ffde297d42ac9dba93f348b644f7.png