【RocketMQ】原理分析：Producer消息发送模式、分发规则

悠悠 2022-12-22 08:38 273阅读 0赞

## 1.消息发送模式 ##

### 1.1 同步发送 ###

同步消息发送模式就是说消息发送出去后，producer会等到broker回应后才能继续发送下一个消息

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

DefaultMQProducer producer = new DefaultMQProducer("my_producer_group");
    producer.send(message); // 同步发送

返回的SendResult有四种状态：SEND OK（成功），FLUSH\_DISK\_TIMEOUT（刷盘超时），FLUSH\_SLAVE\_TIMEOUT（同步超时），SLAVE\_NOT\_AVAILABLE，

### 1.2 异步发送 ###

异步发送是指发送方发出数据后，不等接收方发回响应，接着发送下个数据包的通讯方式。

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

MQ 的异步发送，需要用户实现异步发送回调接口（SendCallback）。消息发送方在发送了一条消息后，不需要等待服务器响应即可返回，进行第二条消息发送。发送方通过回调接口接收服务器响应，并对响应结果进行处理

DefaultMQProducer producer = new DefaultMQProducer("my_producer_group");
    producer.send(msg, new SendCallback() {
            
        @Override // 成功回调
        public void onSuccess(SendResult sendResult) {
            
            System.out.printf("%s%n",sendResult);  
        }    
        @Override // 失败回调
        public void onException(Throwable throwable) {
             
            throwable.printStackTrace();  
        } 
    });

### 1.3 单向发送 ###

单向（Oneway）发送特点为发送方只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答.效率最高

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

DefaultMQProducer producer = new DefaultMQProducer("my_producer_group");
    producer.sendOneway(msg);

### 1.4 顺序发送 ###

在kafka中，消息可以通过自定义分区策略来实现消息的顺序发送，实现原理就是把同一类消息都发送到相同的分区上。

在RocketMQ中，是**基于多个Message Queue来实现类似于kafka的分区效果**。如果一个Topic 要发送和接收的数据量非常大， 需要能支持增加并行处理的机器来提高处理速度，这时候一个Topic 可以根据需求设置一个或多个Message Queue。Topic 有了多个Message Queue 后，消息可以并行地向各个 Message Queue 发送，消费者也可以并行地从多个Message Queue 读取消息并消费。

## 2.分发规则 ##

### 2.1 默认规则 ###

通过自定义发送策略来实现消息只发送到同一个队列，因为一个Topic 会有多个Message Queue ，如果使用Producer 的默认配置，这个Producer 会轮流向各个Message Queue 发送消息。Consumer 在消费消息的时候，会根据负载均衡策略，消费被分配到的Message Queue。

### 2.2 自定义 ###

如果不经过特定的设置，某条消息被发往哪个Message Queue ，被哪个Consumer 消费是未知的。如果业务需要我们把消息发送到指定的Message Queue 里，比如把同一类型的消息都发往相同的 Message Queue。那是不是可以实现顺序消息的功能呢？

和kafka一样，rocketMQ也提供了消息路由的功能，我们可以自定义消息分发策略，通过实现 MessageQueueSelector，来实现自己的消息分发策略

DefaultMQProducer producer = new DefaultMQProducer("my_producer_group");
    SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
            
        @Override   
        public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
           
            int key=o.hashCode();     
            int size = list.size(); // 当前topic的消息队列数
            int index = key%size;   // 目标队列索引   
            return list.get(index); // list.get(0);   
        }
    },"key_"+i);

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/074a00f26a30415ba8bfc76e31881b3c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/320e8b09cb754d5f8085e37412c99d02.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/3b192fbea0814eca81660f3fa5799197.png