RocketMQ实现顺序消息原理分析

电玩女神 2022-10-20 12:13 338阅读 0赞

### 文章目录 ###

*   *   *  （一）顺序消息的使用场景
         *  （二）如何发送和消费顺序消息
         *   *  1. 顺序发送消息
             *  2. 顺序消费消息
         *  （三）顺序发送的技术原理
         *   *  1. 分类
             *  2. 消息发送方式
             *  3. 技术原理
         *  （四）普通发送的技术原理
         *   *  1. 分析
             *  2. 并发消费的技术原理
         *  （五）消息的幂等性

### （一）顺序消息的使用场景 ###

日常中需要保证顺序的应用场景非常多，也就是顺序消息和使用场景非常多。例如交易系统中的订单创建、支付、退款等流程，先创建订单才能支付，支付完成的订单才能退款，这需要保证先进先出（First In First Out，缩写FIFO)。又例如数据库的BinLog消息，数据库执行新增语句、修改语句，BinLog消息的顺序也必须保证是新增消息、修改消息。

### （二）如何发送和消费顺序消息 ###

在介绍之前，我自己先按照正常的方式发送了消息测试，发送的消息如下字符串

List<String> list = Arrays.asList("A00001", "B00001", "C00001", "d00001");

通过下面的截图可以看到消费端是接收的消息不是同步，A-C-B-D的乱序消息

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

#### 1. 顺序发送消息 ####

【1】修改配置文件  
![在这里插入图片描述][20210517212639360.png]

server.port=8081
    spring.cloud.stream.rocketmq.binder.name-server=192.168.208.32:9876
    spring.cloud.stream.bindings.output.destination=TopicTest
    spring.cloud.stream.rocketmq.bindings.output.producer.group=demo-group
    
    
    
    spring.cloud.stream.bindings.userOutput.destination=TopicUser
    spring.cloud.stream.rocketmq.bindings.userOutput.producer.group=user-group
    
    #设置同步发送
    spring.cloud.stream.rocketmq.bindings.output.producer.sync=true

本次发送的Topic是TopicTest

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 1]  
【2】编写顺序发送消息测试代码

/** * 测试顺序发送消息 * @auther zhangyu */
        @GetMapping("/sendSync")
        public  String  sendMsgSync( ){ 
            List<String> list = Arrays.asList("A00001", "B00001", "C00001", "d00001");
            for (String message : list) { 
                MessageBuilder<String> messageBuilder = MessageBuilder.withPayload(message)
                        //设置Header头，表示发送至特定的分区，这个表示发送至第0个队列
                        .setHeader(BinderHeaders.PARTITION_HEADER,0);
                Message<String> mqMessage = messageBuilder.build();
                source.output().send(mqMessage);
            }
            return "消息发送成功";
        }

【3】代码说明  
1：本次顺序发送了"A00001", “B00001”, “C00001”, "d00001"消息  
2：spring.cloud.stream.rocketmq.bindings.output.producer.sync=true表示同步发送，默认就是异步发送  
3：setHeader(BinderHeaders.PARTITION\_HEADER,0); 指定特定的分区队列，表示这是一条顺序消息，将消息固定的发送到第0个消息队列

#### 2. 顺序消费消息 ####

【1】修改配置文件  
![在这里插入图片描述][20210517212719925.png]

server.port=8082
    spring.cloud.stream.rocketmq.binder.name-server=192.168.208.32:9876
    spring.cloud.stream.bindings.input.destination=TopicTest
    spring.cloud.stream.bindings.input.group=input-group
    
    # 默认是并发消费，这里设置为顺序消费
    spring.cloud.stream.rocketmq.bindings.input.consumer.orderly=true

【2】编写测试代码

@SpringBootApplication
    @EnableBinding({ Sink.class,InputChannel.class})
    public class ConsumerApplication { 
    
        public static void main(String[] args) { 
            SpringApplication.run(ConsumerApplication.class, args);
        }
    
    
        /** * 顺序消费消息 */
        @StreamListener(value = Sink.INPUT)
        public void receiveSync(String receiveMsg) { 
            System.out.println("接收消息==》" + receiveMsg+"时间戳==》"+System.currentTimeMillis());
        }
    
    }

1.  运行测试

测试如下端口：http://localhost:8081/sendSync

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 3]  
上图可以看出与我开头消费的顺序不同，本次是按照发送的消息顺序消费

### （三）顺序发送的技术原理 ###

#### 1. 分类 ####

RocketMQ的顺序消息分2种情况:局部有序和全局有序。

【局部有序】  
指发送同一个队列的消息有序，可以在发送消息时指定队列在消费消息时也按顺序消费。上面的演示的案例就是局部有序，这也是顺序发送最常用的情况。

【全局有序】  
默认Topic对应多个队列，当设置Topic只有1个队列可以实现全局有序，创建Topic时手动设置。此类场景极少，性能差，通常不推荐使用。

#### 2. 消息发送方式 ####

RocketMQ中消息发送有三种方式:同步、异步、单向。

同步:发送网络请求后会同步等待Broker服务器的返回结果，支持发送失败重试，适用于较重要的消息通知场景。

异步:异步发送网络请求，不会阻塞当前线程，不支持失败重试，适用于对响应时间要求更高的场

单向:单向发送原理和异步一致，但不支持回调。适用于响应时间非常短  
对可靠性要求不高的场景，例如日志收集。

顺序消息发送的原理很简单，同一类消息发送到相同的队列即可。为了保证先发送的消息先存储到消息队列，必须使用同步发送的方式，否则可能出现先发的消息后到消息队列，此时消息就乱序了。

#### 3. 技术原理 ####

RocketMQ支持两种消息模式︰集群消费(Clustering）和广播消费(Broadcasting)。两者的区别是，在广播消费模式下每条消息会被ConsumerGroup的每个Consumer消费，在集群消费模式下每条消息只会被ConsumerGroup的一个Consumer消费。

多数场景都使用集群消费，消息每消费一次代表一次业务处理，集群消费表示每条消息由业务应用集群中任意一个服务实例来处理。少数场景使用广播消费，例如数据发生变化，更新业务应用集群中每个服务的本地缓存，这就需要一条消息整个集群都消费一次，默认是集群消费，消息模式是前提条件，我们下面也仅分析这种模式下的情况。

顺序消费也称为有序消费，原理是同一个消息队列只允许Consumer中的一个消费线程拉取消费。Consumer中有个消费线程池，多个线程会同时消费消息。在顺序消费的场景下消费线程请求到Broker时会先申请独占锁，获得锁的请求则允许消费。

消息消费成功后，会向Broker提交消费进度，更新消费位点信息，避免下次拉取到已消费的消息。顺序消费中如果消费线程在监听器中进行业务处理时抛出异常，则不会提交消费进度，消费进度会阻塞在当前这条消息，并不会继续消费该队列中后续的消息，从而保证顺序消费。在顺序消费的场景下，特别需要注意对异常的处理，如果重试也失败，会一直阻塞在当前消息，直到超出最大重试次数，从而在很长一段时间内无法消费后续消息造成队列消息堆积。

### （四）普通发送的技术原理 ###

#### 1. 分析 ####

普通消息发送就是并发式的发送和消费消息，这也是我们日常开发过程中最常用的方式，最常用的场景是系统间的异步解耦和流量的削峰填谷，这些场景下尽量保证消息高性能收发即可。

从普通消息与顺序消息的对比来看，普通消息在发送时选择消息队列的策略不同。普通消息发送选择队列有两种机制:轮询机制和故障规避机制（也称故障延迟机制)。默认使用轮询机制，一个Topic有多个队列，轮询选择其中一个队列。

轮询机制的原理是路由信息TopicPublishInfo中维护了一个计数器sendWhichQueue，每发送一次消息需要查询一次路由，计算器就进行“+1”，通过计数器的值index与队列的数量取模计算来实现轮询算法。

#### 2. 并发消费的技术原理 ####

RocketMQ支持两种消费方式:顺序消费和并发消费。并发消费是默认的消费方式，日常开发过程中最常用的方式，除了顺序消费就是并发消费。

并发消费也称为乱序消费，其原理是同一个消息队列提供给Consumer中的多个消费线程拉取消费。Consumer中会维护一个消费线程池，多个消费线程可以并发去同一个消息队列中拉取消息进行消费。如果某个消费线程在监听器中进行业务处理时抛出异常，当前消费线程拉取的消息会进行重试，不影响其他消费线程和消息队列的消费进度，消费成功的线程正常提交消费进度。

并发消费相比顺序消费没有资源争抢上锁的过程，消费消息的速度比顺序消费要快很多。

### （五）消息的幂等性 ###

说到消息消费不得不提消息的幂等性，业务代码中通常收到一条消息进行一次业务逻辑处理，如果一条相同消息被重复收到几次，是否会导致业务重复处理?Consumer能否不重复接收消息

在所有消息系统中消费消息有三种模式:at-most-once(最多一次)、at-least-once(最少一次)和exactly-only-once(精确仅一次），分布式消息系统都是在三者间取平衡前两者是可行的并且被广泛使用的。

at-most-once:消息投递后不论消费是否成功，不会再重复投递，有可能会导致消息未被消费，RocketMQ未使用该方式。

at-least-once:消息投递后，消费完成后，向服务器返回ACK（消费确认机制），没有消费则一定不会返回ACK消息。由于网络异常、客户端重启等原因，服务器未能收到客户端返回的ACK，服务器则会再次投递，这就会导致可能重复消费，RocketMQ通过ACK来确保消息至少被消费次。

exactly-only-once:必须下面两个条件都满足，才能认为消息是“Exactly Only Once”。  
【1】发送消息阶段，不允许发送重复的消息;  
【2】消费消息阶段，不允许消费重复的消息。在分布式系统环境下，如果要实现该模式，巨大的开销不可避免。RocketMQ为了追求高性能，并不保证此特性，无法避免消息重复，由业务上进行幂等性处理。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70]: /images/20221020/b5e3517169c245f6845a21977d1dc38b.png
[20210517212639360.png]: /images/20221020/9d1ffadad7c2440ab4bf4c1d8ea060a3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 1]: /images/20221020/b72f143df05942a285a9ed706e22908f.png
[20210517212719925.png]: /images/20221020/e3839a2ab605447aa05f56b9d737bcb9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 2]: /images/20221020/597140cd2f884f0c843bb73b76bb933e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 3]: /images/20221020/8b98103d08a24e3091cc04c95719f929.png