第二章深入RabbitMQ高级特性.md · wanggonghuitao/RabbitMQ 学习笔记

### 二、深入RabbitMQ的高级特性

#### 1.消息的可靠性投递及解决方案

- 什么是生产端的可靠性投递？
    - 保障消息的成功发出
    - 保障MQ节点的成功接收
    - 发送端收到MQ节点（Broker）确认应答
    - 完善的消息进行补偿机制
- BAT/TMD 互联网大厂的解决方案：
    - 消息落库，对消息状态进行打标
    - 消息的延迟投递，做二次确认，回调检查
- 注意：大厂在高并发时都不会使用事务，事务会很容易造成性能的瓶颈。

#### 2.消息如何保障 100% 的投递成功（方案一）
- 消息信息落库，对消息状态进行打标
    - ![第一种方案图](https://images.gitee.com/uploads/images/2018/1009/212618_481e21f8_2206730.png "屏幕截图.png")
    - 上图蓝色的框就是生产者，实现步骤如下：
        - 1.生产者将要发送的数据信息（假设是订单）落库(BIZ DB)；然后创建一条消息存储到MSG DB，该消息包含状态status和发送消息的时间timeout，默认status为0表示刚投递，如果两个入库都成功，则执行Step2。
        - 2.当订单和消息都成功入库后，将该消息发送到MQ Broker。
        - 3.另起一个线程进行确认监听，监听MQ Broker的收到消息的回复broker confirm。
        - 4.当监听到MQ Broker的确认回复后，将MSG DB中的消息状态status修改为1，表示消息投递成功。如果监听不到MQ Broker的确认，则执行补救措施，即第5步。
        - 5.创建一个分布式定时任务，从MSG DB中查询出status=0并且已经投递超过3分钟的消息。使用分布式定时任务目的是防止在集群架构下多个定时任务同时去获取status=0的消息。
        - 6.将Step5中获取的消息重新发送到MQ Broker，即重复第二步及以后的操作。
        - 7.当某个消息重复投递次数超过3次还没有投递成功，则将消息的状态设置为2，即投递失败，这种情况非常之少，一般都是业务上的错误，需要人工排查。

- 这种方案可以保证消息的可靠性投递，但是不适合在高并发场景下应用
    - 原因：这种方式需要两次消息落库写磁盘，第一次是数据信息落库，第二次是消息信息落库，数据库可能会遇到瓶颈。其实我们只要进行业务数据落库即可，即第二种方案。

#### 3.消息如何保障 100% 的投递成功（方案二）
- 消息的延迟投递，做二次检查，回调检查
    - ![方案二](https://images.gitee.com/uploads/images/2018/1009/230050_723457de_2206730.png "屏幕截图.png")
    - 蓝色的Upstream为上游服务（生产端），Downstream为下游服务（消费端）。绿色的线表示一个队列通信，蓝色的线表示一个队列通信，红色的线表示一个队列通信。实现步骤如下：
        - 1.Upstream service首先一定是先数据信息（假设订单）入库（BIZ DB）。数据入库成功后，根据数据信息创建两个消息。
        - 2.发送Message1到MQ Broker,再延迟一定时间发送Message2（假设延迟3分钟）到MQ Broker。
        - 3.Downstream 监听队列获取 Message1，并对Message1进行相应的处理。
        - 4.当Step3消费Message1后，重新创建一个表示消费过Message1的消息，发送到MQ Broker。
        - 5.Callback service监听Step4发送的确认消费的消息，将收到的消息落库到MSG DB。
        - 6.当Message1发送3分钟以后，延迟检查消息Message2会发送到MQ Broker。此时Callback service监听红线代表的队列获取到Message2消息，然后再去MSG DB中检查是否有Message1对应的消费确认消息。如果有，则表示投递成功。如果没有，则执行Step7。
        - 7.Callback service会主动发起一个RPC通信，告诉上游服务Upstream延迟检查消息没有在MSG DB找到，需要重新发送一遍。这时Callback 会发一个ReSend Command命令，并且带一个id过来，Upstream去BIZ DB中查找数据信息，重新创建两个消息，即又回到了Step2，继续重复执行。
    
- 这种方案主要关注的是性能，通过异步补偿的去解决100%投递，其实Callback就是一个补偿服务（非主流程服务），也达到了解耦的目的。

#### 4.幂等性的概念及业界主流解决方案
- 幂等性是什么？
    - 借鉴数据库乐观锁机制来看一下：
    - 比如说我们要执行一条更新库存的SQL语句：
        ```
        UPDATE goods_store SET count = count - 1 WHERE id = 33 
        ```
    - 在高并发情况下，两个用户同时下订单，导致两个用户都查询到了库存中还有一件商品，然后这两个用户更新库存可能会使商品数量减为负数。
    - 为了避免上面所述的并发问题，可以通过加版本号的方式来解决，SQL如下：
        ``` 
        UPDATE goods_store SET count = count - 1, version = version + 1
        WHERE version = 1 and id = 33
        ```
    - 高并发情况下，两个用户同时下单，这时两个用户查出相同的商品库存数和版本号，接下来用户1去更新库存，库存变为零并且版本号加1，然后用户2去更新库存会发现根据id和版本号找不到对应的库存记录，通过这种方式来保证乐观锁和幂等性。
    - 注意：在防止商品超卖的解决方案中，业务上的控制可能会有万分之一的差错，但是在数据库层进行利用数据库事务去保障是非常可靠的，举例SQL如下。
        ```
        UPDATE goods_store SET count = count - 1, version = version + 1
        WHERE version = 1 and id = 33 and count > 0
        ```    
-  **幂等性的概念** ：无论一个操作执行多少次，其结果总是相同的。

-  **消费端-幂等性保障** 
    - 在海量订单产生的业务高峰期，如何避免消息的重复消费问题？
        - 消费端实现幂等性，就意味着我们的消息永远不会重复消费多次，即使我们收到了多条一样的消息。
    - 业界主流幂等性操作：（从技术的角度来看有下面两种方式）
        - **唯一ID + 指纹码 机制，利用数据库去重。**
        - **利用Redis的原子性实现。**
    - 消费端幂等性保障（方案一）：唯一ID + 指纹码 机制
        - 通常我们会为消息生成一个全局唯一ID，但这往往不够，需要加上指纹码。
            - 原因：因为有的用户可能在一瞬间频繁的进行了多次消费，比如用户转账100元，在这一瞬间又转了100元，这时就可能是两种情况：第一种，用户只想转账一次，但是由于其他原因导致我们接收到了两次转账请求；第二种就是用户真的想转两次100元。此时就需要加上指纹码来生成数据库ID。
            - **指纹码**：指纹码可能是我们的一些业务规则或者是时间戳加上具体的银行返回的一个唯一的信息码，它并不一定是我们系统去生成的，也可是由外部的规则或内部的业务规则去拼接起来的一个东西。 它的目的就是为了保证这次操作是**绝对唯一**的。
            - 通过 全局ID + 指纹码 = ID，保证ID在数据库里是一个唯一的主键，利用数据库主键去重，来防止消息被消费。
        - 收到消息后，首先在数据库中查询消息是否被消费过，SQL如下：
            ```
                -- count(1) 只查询主键 
                SELECT COUNT(1) FROM goods_store WHERE ID = 唯一ID + 指纹码 
            ```
            - 查询结果是1，表示消息已被消费，不处理消息，返回失败。
            - 查询结果是0，表示消息没被消费，处理消息。
        - 好处：实现简单。
        - 坏处：高并发下有数据库写入的性能瓶颈。
        - 解决方案：跟进ID进行分库分表，利用路由算法进行分压分流。
       
    - 消费端幂等性保障（方案二）：利用Redis原子特性实现
        - 利用Redis的原子特性，无论是在单点或集群情况下：都可以通过set一个key，然后由消息过来的时候判断这个key是否存在，从而确定消息有没有被消费；或者通过自增key的方式获取此key的值来判断消息有没有被消费。
        - 使用Redis实现幂等，需要考虑的问题：
            - 第一：我们是否要对数据进行落库，如果落库的话，关键解决的问题是：数据库和缓存如何做到原子性？即如何做到同时成功或同时失败？ Redis是内存数据库，性能超强。
                - 一般策略是先写数据库，再删缓存。不过我们要每次查redis，因此删完缓存之后再查询一次数据库放入redis中。
            - 第二：如果不进行落库，那么都存储在缓存中，如何设置定时同步策略？缓存如果出现问题，怎么办？
        - 考虑清楚了上述问题，再使用这种方案。
    
#### 5.Confirm确认消息详解
- 理解Confirm消息确认机制：
    - 消息的确认，是指生产者投递消息后，如果Broker收到消息，则会给我们生产者一个应答。
    - 生产者进行接收应答，用来确定这条消息是否正常的发送到Broker，这种方式也是消息可靠性投递的核心保障。
- 确认机制流程图
    - ![确认机制流程图](https://images.gitee.com/uploads/images/2018/1014/101306_1f032cfd_2206730.png "屏幕截图.png")
- Confirm确认消息如何实现(在生产端配置)
    - 第一步：在channel上开启确认模式：channel.confirmSelect()
    - 第二步：在channel上添加监听：**addConfirmListener**，监听成功和失败的返回结果，根据具体的结果对消息进行重新发送、或记录日志等待后续处理，监听如下：
        ```
        //6 添加一个确认监听
        channel.addConfirmListener(new ConfirmListener() {
            // ack 时执行该方法
            @Override
            public void handleAck(long l, boolean b) throws IOException {
                System.err.println("---------------Ack!---------------");
            }
            // 失败时执行该方法
            @Override
            public void handleNack(long l, boolean b) throws IOException {
                System.err.println("---------------no ack!--------------");
            }
        });

```  
- Confirm消息确认可能会有三种情况：
    - 第一：收到ack确认回复，执行正确的业务逻辑
    - 第二：收到了Nack失败的确认回复，那可能是磁盘满了、RabbitMQ异常、queue容量满了等原因。
    - 第三：我们没有收到回复，我们不知道消息是否投递成功或者Broker是否接收成功并返回确认消息，因此要结合前面的消息100%投递去实现。

#### 6.Return返回消息详解（在生产端配置）
- Return Listener用于处理一些不可路由的消息！
- 我们消息的生产者，通过指定一个路由Exchange 和 RoutingKey，把消息送达到某一个消息队列中去，然后我们的消费者监听队列，进行消费处理操作。
- 但是在某些情况下，如果我们在发送消息的时候，当前的Exchange不存在或者指定的路由Key路由不到，这个时候如果我们需要监听这种不可达的消息，就要使用Return Listener!
- 在基础 API中有一个关键的配置项，在：
    - Mandatory：如果为true，则监听器会接收到路由不可达的消息，然后进行后续处理，如果为false，那么Broker端会自动删除该消息。（默认是false）
    - Mandatory设置为true以后，在配置一个Return Listener即可，如下：
        ```
        channel.addReturnListener(new ReturnListener() {
            /**
             *
             * @param i 响应码
             * @param s 响应文本
             * @param s1 路由的交换机
             * @param s2 路由的key
             * @param basicProperties
             * @param bytes 消息体内容
             * @throws IOException
             */
             @Override
             public void handleReturn(int i, String s, String s1, String s2,
                                     AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException {
                System.err.println("---------handler return--------");
                System.err.println("replyCode: " + i);
                System.err.println("replyText: " + s);
                System.err.println("exchange: " + s1);
                System.err.println("queue: " + s2);
                System.err.println("basicProperties: " + basicProperties);
                System.err.println("body: " + new String(bytes));
            }
        });
                    
        ```
- return 消息机制流程
    - ![return 消息机制流程](https://images.gitee.com/uploads/images/2018/1014/131845_e7187bed_2206730.png "屏幕截图.png")

#### 7.自定义消费者使用
- 我们一般就是在代码中编写while循环，进行Consumer.nextDelivery方法进行获取下一条消息，然后进行消费处理！
- 但是我们使用自定义的Consumer更加的方便，解耦性也更强，也是在实际工作中最常用的使用方式。
- 如何实现消费端的自定义监听？
    ```
    public class MyConsumer extends DefaultConsumer{

public MyConsumer(Channel channel) {
            super(channel);
        }

@Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                               byte[] body) throws IOException {
            System.err.println("-------------consumer message-------------");
            System.err.println("body：" + new String(body));
        }
    }
    ```

#### 8.消费端的限流策略
- 什么是消费端的限流：
    - 假设一个场景，首先，我们RabbitMQ服务器上有上万条未处理的消息，我们随便打开一个消费者客户端，会出现下面情况：
    - 巨量的消息瞬间全部推过来，但是我们单客户端无法同时处理这么多数据。消费端服务端可能会卡顿或卡死的。
- RabbitMQ 提供了一种 qos（服务质量保证）功能，即在**非自动确认消息**（autoAck=false）的前提下，如果一定数目的消息（通过基于consume或者channel设置Qos的值）未被确认，则不进行消费新的消息。
- 有关限流的方法：void BasicQos(uint prefetchSize, ushort prefetchCount, bool global);
    - prefetchSize:0，表示单条消息的大小限制，0表示不限制。
    - prefetchCount：会告诉RabbitMQ不要同时给消费者推送多于N个消息。即一旦有N个消息还没有ack，则该consumer将block掉，直到有消息ack。 开发中一般设置为1，即消息没有被确认，就不接收新消息。
    - global：true/false 是否将上面的消息设置应用为channel。简单来说，就是限制是cahnnel级别还是consumer级别
    - 注意：
        - prefetchSize 和 global这两项，rabbitmq没有实现，提供的机制是prefetchSize=0，global=false，暂不研究。
        - prefetchCount在no_ack=false的情况下生效，即在自动应答的情况下这两个值是不生效的。
- 实现方式：
    - 消费端：
        ```
        /** 消费端限流: 首先是将 autoAck设置为false */
        channel.basicQos(0, 1, false);
        channel.basicConsume(queueName, false, new MyConsumer(channel));
        ```
    - 消费者：
        ```
        public class MyConsumer extends DefaultConsumer{

private Channel channel;
            public MyConsumer(Channel channel) {
                super(channel);
                this.channel = channel;
            }

@Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                               byte[] body) throws IOException {
                // 手动签收, 第二个参数表示是否批量签收
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        }
        ```

#### 9.消费端ACK与重回队列机制
- 消费端的手工ACK与NACK
    - 消费端进行消费的时候，如果由于业务异常我们可以进行日志的记录，然后进行补偿。
    - 如果由于服务器端宕机等严重问题，那我们就需要手工ACK 保障消费端消费成功。比如消费端消费完了消息，但是由于宕机导致不能ack，此时就需要手工ACK；但是如果消费了一半消息，宕机了，那么就不需要手工ACK。

- 消费端的重回队列
    - 消费端重回队列是对于没有处理成功的消息，把消息重新投递给Broker，并且放在队尾。
    - 一般在实际应用中，都会关闭重回队列，也就是设置为False
- 实现如下：
    - 生产端：
    ```
        for(int i=0; i < 5; i++){

Map<String, Object> headers = new HashMap<>();
            headers.put("num", i);
            AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
                    .deliveryMode(2)
                    .contentEncoding("UTF-8")
                    .headers(headers)
                    .build();

String msg = "Hello RabbitMQ ack message " + i;
            channel.basicPublish(exchangeName, routingKey, false, properties, msg.getBytes());
        }    
    ```
    - 消费端
    ```
        /** 手工签收 必须要关闭 autoAck = false */
        channel.basicConsume(queueName, false, new MyConsumer(channel));
    ```
    - 消费者
    ```
        if ((Integer)properties.getHeaders().get("num") == 0){
            // 第三个参数表示 是否让消息重回Broker端队列的尾部
            channel.basicNack(envelope.getDeliveryTag(), false, true);
        }else{
            // 手动签收, 第二个参数表示是否批量签收
            channel.basicAck(envelope.getDeliveryTag(), false);
        }
    ```

#### 10.TTL消息/队列
- TTL
    - TTL是Time To Live的缩写，也就是生存时间。
    - RabbitMQ 支持消息的过期时间，在消息发送时可以指定，发送消息时在properties参数中设置消息的过期时间。
    - RabbitMQ 支持队列的过期时间，从消息进入队列开始计算，只要超过了队列的超时时间配置，那么消息会自动的清除。

#### 11.死信队列详解 - 1
- **死信队列**：DLX，Dead-Letter-Exchange
    - 利用死信队列，当一个消息在队列中变成死信（dead message）之后，他能被重新publish到另一个Exchange，这个Exchange就是DLX。
    
- 消息变成死信有下面几种情况
    - 消息被拒绝（basic.reject/basic.nack）并且requeue=false。  
    - 消息TTL过期。
    - 队列达到最大长度。

- 死信队列使用方式
    - DLX也是一个正常的Exchange，和一般的Exchange没有什么区别，它能在任何队列上被指定，实际上就是设置某个队列的属性。
    - 当队列中有死信时，RabbitMQ就会自动将这个消息重新发布到设置的Exchange上去，进而被路由到另一个队列。
    - 可以监听这个队列做相应的处理，这个特性可以弥补RabbitMQ3.0以前支持的immediate参数的功能。

- 死信队列设置
    - 首先需要设置死信队列的 exchange和queue，然后进行绑定：
        - Exchange: dlx.exchange
        - Queue：dlx.queue
        - RoutingKey: #
    - 然后我们进行正常声明交换机、队列、绑定，只不过我们需要在队列上加一个参数即可：arguments.put("x-dead-letter-exchange","dlx.exchange");
    - 这样的消息在过期、被拒绝并且requeue=false、队列达到最大长度时、消息就可以直接路由到死信队列。

#### 12.死信队列详解 - 2

wanggonghuitao/RabbitMQ 学习笔记

简介

发行版

贡献者

近期动态

wanggonghuitao/RabbitMQ 学习笔记 .gitee-modal { width: 500px !important; }

简介

发行版

贡献者

近期动态

搜索帮助

wanggonghuitao/RabbitMQ 学习笔记