RocketMQ生产者核心配置和核心知识讲解

约定不等于承诺〃 2023-07-04 15:36 94阅读 0赞

# 1. 消息队列RocketMQ生产者核心配置 #

生产者常见核心配置：

1.  compressMsgBodyOverHowmuch：消息超过默认字节4096后进行压缩
2.  retryTimesWhenSendFailed：失败重复次数
3.  maxMessageSize：最大消息配置，默认128k
4.  topicQueueNums：主题下面的队列数量，默认是4
5.  autoCreateTopicEnable：是否自动创建topic，开发建议为True，生产要为false
6.  defaultTopicQueueNums：自动创建服务器不存在的topic，默认创建的队列数
7.  autoCreateSubscriptionGroup：是否允许Broker自动创建订阅组，建议线下开发开启，线上关闭
8.  brokerClusterName：集群名称
9.  brokerId：0表示Master主节点，大于0表示从节点
10. brokerIP ：broker服务地址
11. brokerRole：broker角色 ASYNC\_MASTER异步复制/ SYNC\_MASTER同步复制/ SLAVE无影响
12. deleteWhen：每天执行删除过期文件的时间，默认每天凌晨4点
13. flushDiskType：刷盘策略，默认为 ASYNC\_FLUSH(异步刷盘)，另外是SYNC\_FLUSH(同步刷盘)
14. listenPort：Broker监听的端口号
15. mapedFileSizeCommitLog：单个commitlog文件大小，默认是1GB
16. mapedFileSizeConsumeQueue：ConsumeQueue每个文件默认存30W条，可以根据项目调整
17. storePathRootDir：存储消息以及一些配置信息的根目录 默认为用户的$\{HOME\}/store
18. storePathCommitLog：commitlog存储目录默认为$\{storePathRootDir\}/commitlog
19. storePathIndex：消息索引存储路径
20. syncFlushTimeout：同步刷盘超时时间
21. diskMaxUsedSpaceRatio：监测可用的磁盘空间大小，超过后会写入报错

# 2. RocketMQ消息发送状态 #

*  消息发送有同步和异步
 *  Broker消息投递状态讲解  
    1. FLUSH\_DISK\_TIMEOUT  
    没有在规定时间内完成刷盘(刷盘策略需要为SYNC\_FLUSH才会出现这个错误)  
    2. FLUSH\_SLAVE\_TIMEOUT  
    主从模式下，broker是SYNC\_MASTER，没有在规定时间内完成主从同步  
    3. SLAVE\_NOT\_AVAILABLE  
    主从模式下，broker是SYNC\_MASTER，但是没有找到被配置成Slave的Broker  
    4. SEND\_OK  
    发送成功，没有发生上面的三种问题  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bqc2RzZw_size_16_color_FFFFFF_t_70]  
    ![在这里插入图片描述][20200212124738536.png]

# 3. RocketMQ生产和消费消息重试及处理 #

![在这里插入图片描述][20200212134712263.png]  
从Producer投递消息到Broker，在这个过程中由于网络问题，不能将消息投递到Broker，消息投递失败。或者消息消费时，出现异常，导致消息消费失败。所以需要生产和消费消息重试。

*  生产者Producer重试  
    1. 消息重投(保证数据的高可靠性)，本身内部支持重试，默认次数是2  
    2. 如果网络情况比较差，或者跨集群则建议改多几次
 *  消费端重试  
    1. 原因：消息处理异常、broker端到consumer端各种问题，如果网络原因闪断，消息处理失败，ACK返回失败等等问题。  
    2. 注意：  
    \- 重试问题隔时间配置，默认每条消息最多重试16次  
    messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h  
    \- 超过重试次数人工补偿  
    \- 消费端去重  
    \- 一条消息无论重试多少次、这些重试消息的MessageID,key不会改变  
    \- 消费重试只针对集群消费方式生效；广播方式不提供失败重试特效，即消费失败后，失败消息不再重试，继续消费新的消息

# 4. RocketMQ异步发送消息和回调实战 #

官方文档实例：[https://rocketmq.apache.org/docs/simple-example/][https_rocketmq.apache.org_docs_simple-example]

producer.send(message, new SendCallback(){
     onSuccess(){}
     onException(){}
    })

注意：官方例子：如果异步发送消息，调用producer.shutdown()后会失败  
异步发送：不会重试，发送总次数等于1  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bqc2RzZw_size_16_color_FFFFFF_t_70 1]

# 5. RocketMQ发送消息及多种应用场景对比 #

*  SYNC：同步  
    应用场景：重要邮件通知、报名短信通知、营销短息系统等
 *  ASYNC：异步  
    应用场景：对RT时间敏感，可以支持更高的并发，回调成功触发相对应的业务，比如注册成功后通知积分系统发放优惠券
 *  ONEWAY：无需要等待响应  
    使用场景：主要是日志收集，适用于某些耗时非常短，但对可靠性要求并不高的场景，也就是LogServer，只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答

producer.getProducer().sendOneway(message);

汇总对比：

<table> 
 <thead> 
  <tr> 
   <th>发送方式</th> 
   <th>发送TPS</th> 
   <th>发送结果返回</th> 
   <th>可靠性</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>同步发送</td> 
   <td>快</td> 
   <td>有</td> 
   <td>不丢失</td> 
  </tr> 
  <tr> 
   <td>异步发送</td> 
   <td>快</td> 
   <td>有</td> 
   <td>不丢失</td> 
  </tr> 
  <tr> 
   <td>单向发送</td> 
   <td>最快</td> 
   <td>无</td> 
   <td>可能丢失</td> 
  </tr> 
 </tbody> 
</table>

# 6. RocketMQ延迟消息和电商系统中的应用 #

## 6.1 什么是延迟消息 ##

*  Producer将消息发送到消息队列RocketMQ服务端，但并不期望这条消息立马投递，而是推迟到在当前时间点之后的某一个时间投递到Consumer进行消费，该消息即定时消息，目前支持固定精度的消息
 *  代码：rocketmq-store > MessgeStoreConfig.java 属性 messageDelayLevel

"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

*  使用message.setDelayTimeLevel(xxx)设置延时投递 // xxx是级别，1表示配置里面的第一个级别，2表示第二个级别
 *  除了延迟消息还有**定时消息**，目前rocketmq开源版还不支持，商业版则有，两者使用场景类似  
    ![在这里插入图片描述][20200212220743596.png]

## 6.2 使用场景 ##

*  通过消息触发一些定时任务，比如在某一固定时间点向用户发送提醒消息
 *  消息生产和消费有时间窗口要求：比如在天猫电商交易中超时未支付关闭订单的场景，在订单创建时会发送一条延时消息。这条消息将会在30分钟以后投递给消费者，消费者收到此消息后需要判断对应的订单是否已完成支付。如支付未完成，则关闭订单。如已完成支付则忽略

# 7. RocketMQ生产者之MessageQueueSelector实战 #

简介：生产消息使用MessageQueueSelector投递到Topic下指定的queue，一个topic下默认有4个queue队列。  
应用场景：顺序消费消息，分摊负载  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bqc2RzZw_size_16_color_FFFFFF_t_70 2]

支持同步和异步发送

同步方法 ： public SendResult send(Message msg, MessageQueueSelector selector, Object arg)
    异步方法 ： public void send(Message msg, MessageQueueSelector selector, Object arg, SendCallback sendCallback)

选择的queue数量必须小于配置的，否则会出错

# 8. 讲解顺序消息在电商和证券系统中应用场景 #

什么是顺序消息：**消息的生产和消费顺序一致**

**全局顺序**：topic下面全部消息顺序一致(少用)

*  性能要求不高，所有的消息严格按照FIFO原则进行消息发布和消费的场景，并行度成为消息系统的瓶颈，吞吐量不够高
 *  在证券处理中，以人民币兑换美元为例子，在价格相同的情况下，先出价者优先处理，则可以通过全局顺序的方式按照FIFO的方式进行发布和消费

**局部顺序**：只要保证一组消息被顺序消费即可(RocketMQ使用)

*  性能要求高
 *  电商的订单创建，同一个订单相关的创建订单消息、订单支付消息、订单退款消息、订单物流消息、订单交易成功消息都会按照先后顺序来发布和消费  
    (阿里巴巴集团内部电商系统均使用局部顺序消息，既保证业务顺序，同时又能保证业务的高性能)

**顺序发布**：对于指定的一个Topic，客户端将按照一定的先后顺序发送消息  
**顺序消费**：对于指定的一个Topic，按照一定的先后顺序接收消息，即先发送的消息一定会先被客户端接收到  
**注意**：

*  顺序消息暂时不支持广播模式
 *  顺序消息不支持异步发送方式，否则将无法严格保证顺序

# 9. RocketMQ顺序消费讲解 #

**生产端保证发送消息有序，且发送到同一个队列的同个queue里面**，同一条queue里面，RocketMQ的确是能保证FIFO的  
例子：订单的顺序流程是：创建、付款、物流、完成，订单号相同的消息会被先后发送到一个队列中，  
根据MessageQueueSelector里面自定义策略，根据同个业务id放置到同个queue里面，如订单号取模运算再放到selector中，同一个模的值都会投递到同一条queue

public MessageQueue select(List<MessageQueue> mqs, Message msg, Object
    arg) {
     //如果订单号是字符串，则进行hash，得到一个hash值
     Long id = (Long) arg;
     long index = id % mqs.size();
     return mqs.get((int)index);
     }

消费端要在保证消费同个topic里的同个队列，不应该用MessageListenerConcurrently, 应该使用MessageListenerOrderly，自带单线程消费消息，不能再Consumer端再使用多线程去消费，消费端分配到queue数量是固定的，集群消费会锁住当前正在消费的队列集合的消息，所以会保证顺序消费

# 10. RocketMQ顺序消息生产者投递实战 #

package com.pj.boot.bean;
    
    import lombok.Data;
    
    import java.io.Serializable;
    import java.util.ArrayList;
    import java.util.List;
    
    @Data
    public class ProductOrder implements Serializable { 
    
        private long orderId;
    
        private String type;
    
        public ProductOrder(){ }
    
        public ProductOrder(long orderId, String type) { 
            this.orderId = orderId;
            this.type = type;
        }
    
        public static List<ProductOrder> getOrderList() { 
            List<ProductOrder> list = new ArrayList<>();
            list.add(new ProductOrder(111L, "创建订单"));
            list.add(new ProductOrder(222L, "创建订单"));
            list.add(new ProductOrder(111L, "支付订单"));
            list.add(new ProductOrder(222L, "支付订单"));
            list.add(new ProductOrder(111L, "完成订单"));
            list.add(new ProductOrder(333L, "创建订单"));
            list.add(new ProductOrder(222L, "完成订单"));
            list.add(new ProductOrder(333L, "支付订单"));
            list.add(new ProductOrder(333L, "完成订单"));
            return list;
        }
    
        @Override
        public String toString() { 
            return "ProductOrder{" +
                    "orderId=" + orderId +
                    ", type='" + type + '\'' +
                    '}';
        }
    }

List<ProductOrder> list = ProductOrder.getOrderList();
    
            for (int i = 0; i < list.size(); i++) { 
                ProductOrder order = list.get(i);
                // 生产时建议再加一个key值
                Message message = new Message(JmsConfig.ORDERLY_TOPIC,"taga",order.getOrderId()+"", order.toString().getBytes());
                SendResult sendResult = payProducer.getProducer().send(message, new MessageQueueSelector() { 
                    @Override
                    public MessageQueue select(List<MessageQueue> list, Message message, Object o) { 
                        // 使同一个订单发送到同一个queue里头去
                        Long id = (Long) o;
                        long index = id % list.size();
                        return list.get((int) index);
                    }
                }, order.getOrderId());
                System.out.println(sendResult);
            }

# 11. RocketMQ顺序消息消费者实战 #

*  MessageListenerConcurrently：并发消费
 *  MessageListenerOrderly：顺序消费  
    MessageListenerOrderly并不是简单禁止并发处理，而是为每个Consumer Queue加个锁，消费每个消息前，需要获得这个消息所在的Queue的锁，这样同个时间，同个Queue的消息不被并发消费，但是不同Queue的消息可以并发处理

package com.pj.boot.jms;
    
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.*;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.springframework.stereotype.Component;
    
    import java.io.UnsupportedEncodingException;
    import java.util.List;
    
    @Component
    public class PayOrderlyConsumer { 
    
        private DefaultMQPushConsumer consumer;
    
        private String consumerGroup = "pay_orderly_consumer_group";
    
        public  PayOrderlyConsumer() throws MQClientException { 
    
            consumer = new DefaultMQPushConsumer(consumerGroup);
            consumer.setNamesrvAddr(JmsConfig.NAME_SERVER);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
    
            consumer.subscribe(JmsConfig.ORDERLY_TOPIC, "*");
    
    
            consumer.registerMessageListener(new MessageListenerOrderly() { 
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) { 
                    MessageExt msg = list.get(0);
                    try { 
                        System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msg.getBody()));
                        String topic = msg.getTopic();
                        String body = new String(msg.getBody(), "utf-8");
                        String tags = msg.getTags();
                        String keys = msg.getKeys();
                        System.out.println("topic=" + topic + ", tags=" + tags + ", keys=" + keys + ", msg=" + body);
                        return ConsumeOrderlyStatus.SUCCESS;
                    } catch (Exception e) { 
                        e.printStackTrace();
                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                    }
                }
            });
    
            consumer.start();
            System.out.println("consumer start ...");
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bqc2RzZw_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2020021212472613.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bqc2RzZw==,size_16,color_FFFFFF,t_70
[20200212124738536.png]: https://img-blog.csdnimg.cn/20200212124738536.png
[20200212134712263.png]: https://img-blog.csdnimg.cn/20200212134712263.png
[https_rocketmq.apache.org_docs_simple-example]: https://rocketmq.apache.org/docs/simple-example/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bqc2RzZw_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200212175445261.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bqc2RzZw==,size_16,color_FFFFFF,t_70
[20200212220743596.png]: https://img-blog.csdnimg.cn/20200212220743596.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bqc2RzZw_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200213155039542.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bqc2RzZw==,size_16,color_FFFFFF,t_70