RocketMQ与SpringBoot整合进行生产级二次封装

2022-07-17 14:47:48

零、文章前言说明

前置掌握:SpringBoot基础使用、RocketMQ和SpringBoot的整合使用

  • 主要使用参考第二节
  • 核心使用参考第一篇文章

文章难度:四颗星

代码不难,重点是封装的思想需要体会

不同观点欢迎大家在评论区一起讨论学习,没有对错之分,每个系统业务特性不同,适合系统的才是最好的~

源码地址https://gitee.com/tianxincoder/practice-rocketmq-enterprise
文章只会说明核心代码,其他的基础整合配置和多环境自动隔离参考源码即可

image.png

一、为什么要二次封装

为了不产生歧义,文章中提到的二次封装均是基于原始使用方式的封装,而非源码级别的二次封装

换句话说:如果都需要对源码进行封装了,那么说明公司业务规模都到一定程度了,二次封装这种东西已经不需要讨论了,封装已经是一个共识

  1. 首先明确一点:不进行二次封装完全不影响RocketMQ的使用,可以选择二次封装和不选择二次封装
    • 二次封装可以提供更多的功能和更简洁的使用方式
    • 如果一个封装搞得比原始使用方式更复杂,那么就失去了二次封装的意义
    • Q1:二次封装可不可以不要?完全可以,完全不影响正常使用
    • Q2:二次封装有没有必要?仁者见仁智者见智的问题,如果觉得没有必要那么这篇文章可以跳过~
  2. ORM框架中典型的一个二次封装框架就是MyBatisPlus(简称MP),后者是对MyBatis原生使用的增加,不使用MP直接使用MyBatis可不可以?完全可以,那为什么要用二次封装后的MP?
    • 场景:大部分的数据库操作,无外乎CRUD,那么最常用的比如(根据名称就可以知道这个方法做什么用,就没有必要再二次说明了):updateById、batchUpdate、deleteById、saveOrUpdate、batchInsert。对于上面这5个操作,变化的只是表字段和表名,剩下的语法都是一样
    • 不封装:直接使用MyBatis完全可以自己实现上面方法的功能,但是每个表都需要写一遍自己上面的方法,假设有100张表,那么就会多出495个(下面说明)重复功能代码,而且所有代码都是冗余的
    • 封装后:由封装者提供上面5个方法的公共实现,然后所有需要使用上面功能的Service只需要继承封装好的类就自然的拥有了上面的5大功能,那么代码的冗余量就从100张表*5个方法==500,去掉封装的5个,节省了495的代码冗余量
  3. 所以二次封装是为了更方便用、更简洁、更加适用于系统,量身打造可以大大提升开发效率。就如上面的5个方法,完全重复性的东西为什么要浪费开发时间来做这些冗余的事情呢?

1.1 二次封装不同观点

让我们以一个生活中的蛋炒饭开个头

原始框架好比提供了原材料:厨具、鸡蛋,米饭等食材、菜谱

  1. 对于框架的使用通常有以下两种方式
  • 第一种:根据菜谱来进行做饭(使用原生的方法调用),洗菜、做饭、刷碗、打扫不管啥入参自己管
  • 第二种:找一个人来学会这道菜(负责二次封装的人)的多种做法(封装大部分业务场景)并做成一种点餐式的服务,谁想吃哪种类型的蛋炒饭直接点餐(调用封装好的方法)就可以吃上香喷喷的蛋炒饭

问题:哪种方案更好?
答案:两种各有各的优势(在说废话,哈哈~)

  1. 第一种:原生方式
    • 优点:可以按照各种方式灵活调用,比如每个人都使用RocketMQTemplate原生send发送消息,想要发送什么类型的消息就发送什么类型的消息,比较随意
    • 缺点:代码大量冗余,从构建参数对象、发送消息、消费消息、异常处理、日志记录、异常重试啥啥的都是自己搞,每个消费队列就会出现上面所有的步骤。比如现在有一个订单处理中心A接收来自各种类型订单,此时如B、C两个原始订单来源想让A处理订单,那么B和C都需要按照A的要求进行调用,代码会冗余
  2. 第二种:点餐式服务
    • 优点:封装了大部分统一的方法调用,比如 发送消息、异常处理、日志记录等等都是重复的,封装后点餐的人不需要再关系这一部分要怎么处理,只需要告诉点餐服务要不要进行异常、要不异常重试等等,那么此时对于点餐人来说只需要付钱(调用服务)吃饭(消费消息),除此之外啥也不用管,全部由点餐服务提供者完成所有上述两个步骤外的其它操作
    • 缺点:无法满足所有点餐人的要求,有的人喜欢味道重一点,有的喜欢味道淡一点。但是这个缺点完全可以处理,比如点餐服务提供了自定义厨房(返回原始发送对象),此时调用者可以按照第一种方式进行使用
  3. 选哪种?
    • 个人而言:业务系统复杂的优先选择第二种简单业务的选择第一种(尽量采用封装,后续维护方便)。对于一个复杂的系统,本身业务级的代码就已经很多了,结果还要每个人处理全部一样的东西,消费者越多代码冗余越多。如果一个系统只是为了使用MQ来进行业务分离,消费者也不多,那么可以选择最快的方式,但是最终会选择第二种,如果业务随着时间增长越复杂,越晚改成第二种花费的代价越大!
    • 第一种就好比此时我们要直接操作内存,原生操作就好比C++或C,可以直接操作内存,但是同时用完后还要自己写各种异常处理和释放内存;代码封装就好比Java,我们只需要告诉Java我们要使用内存,然后用完就不用管
  4. 企业中,业务功能产出是一级优先级,在此之上才能有更高级的东西。技术服务于业务,而不是业务服务于技术!比如现在30个人的系统,我们要使用缓存加速访问,那么我们是选择 内部缓存(直接用集合或者map存起来)还是用Redis?
    • 内部缓存和Redis能不能达到目的?能
    • 哪个更方便更快?内部缓存!内部对象就很快实现
    • 如果业务发展迟早会转为Redis这种专业的缓存中间件,就好比业务发展前第一种,业务发展后选择第二种,但是对于大部分业务系统来说功能增加是很快的,特别是产品同事上一分钟提需求下一分钟就要上线这种(开个玩笑~),所以我们在引用一个技术需不需要进行二次封装时需要技术负责人对业务增长有一个预判。建议是都进行封装一下

1.2 封装的抽离点

  1. 对于二次封装,其中最主要的就是找出该框架在日常使用中所出现的大部分涉及到的操作,然后找出变化操作和不变化操作
  2. RocketMQ日常使用主要场景为例:
    • 发送消息阶段:准备需要发送的消息、发送消息、记录原始消息日志、发送失败处理、可靠性处理
    • 消费消息阶段:记录接收消息日志、业务处理、业务日志记录、异常处理、异常重试、异常通知、死信处理
  3. 提取变化点和不变化点(可以抽取为公共处理的场景)
    • 发送消息阶段:
      • 变化点:准备需要发送的消息
      • 不变化点:发送消息、记录原始消息日志、发送失败处理、可靠性处理
    • 消费消息阶段
      • 变化点:业务处理、业务日志记录
      • 不变化点:记录接收消息日志、异常处理、异常重试、异常通知、死信处理
  4. 从上可以看到,对于RocketMQ的使用,大部分场景都是可以抽离成一个公共的方法处理,只有业务级的需要自己处理,所以如果我们把不变化场景抽取后,每个同事只需要写自己业务相关部分即可
  5. 抽取后的复杂度:对于新加一个消费者,只需要处理业务相关三个场景(准备需要发送的消息、业务处理、业务日志记录),剩下的九个场景,只需要封装一次就可以。需要现在就几十个消费者,可以想想一些减少了多少代码冗余

1.3 设计模式的应用

  1. 要封装出一个好的抽象层,【设计模式】建议好好体会和学习一下
  2. 设计模式对于用不到的人来说比较虚幻,对于用的到的人来说,这个真牛X

二、二次封装核心要点

2.1 二次封装核心点

2.1.1 封装主要讨论点

  1. 对于RocketMQ或者说对于整个MQ体系来说(不管是RabbitMQ、RocketMQ、Kafka)等封装的核心主要有两个:发送消息、消费消息者两个场景
  2. 对于RocketMQ我们主要讨论三个地方:RocketMQTemplate封装、RocketMQListener封装和广播消息的封装
  3. 广播消息是分布式系统中同时让所有节点都干一件事情的一个好的方式,如果用不到忽略广播消息即可

2.1.2 发送/消费的几种消息实体

  1. RocketMQ发送消息对于不同的使用来说,大部分选择下面的几种发送消息类型
    • A、发送Json对象,比如Fastjson的JSONObject
    • B、直接发送转Json后的String对象
    • C、根据业务封装对应实体类
    • D、直接使用原生MessageExt接收
  2. 怎么选择?怎么选择才是最优?
    • 上面哪一种都可以达到目的,如果要统一封装就必须要有一个标准
    • 怎么选择只需要回答这个问题:在不看消息发送者的情况下,消费者怎么知道发送者发送的消息含义?
    • 比如现在有一个订单消息,如果我们不看消息发送者,怎么知道发送者给消费者发送哪些字段
      • A、B、D可以吗?一定不可以!JSON对象和String对象,如果我们不看消息发送者不可能知道到底发送了啥,这点我相信没有可以讨论的地方,因为类型决定了这个操作不可能
      • C可以吗?可以!此时不需要看消息发送者,只需要看消费者的实体类点进去,有哪些业务字段一清二楚
      • 可能有杠要抬了,有看实体类的功夫,我看消息发送者都看完了
        • 灵魂拷问1:如果消息发送者和消费者不在一个系统怎么看?邪魅一笑,不同业务线可能没代码权限吧?分布式系统完全独立可能吧?
        • 灵魂拷问2:如果现在需要一个功能,如果某些必须要的字段消息发送者如果没有给的话需要校验,普通String和JSONObject怎么实现?换成实体类呢?
  3. 基于上述讨论点,封装建议基于实体类来,实体类不管是排查问题、新人熟悉系统代码、信息校验等String和JSONObject无法像实体类一样轻松胜任

2.2 RocketMQTemplate封装

2.2.1 封装基础实体类

  1. 基础消息实体类封装了除了业务消息外所有其他公共字段,主要看下面代码中的字段和注释
  2. 基础抽象消息实体,包含基础的消息、根据自己的业务消息设置更多的字段
    • 其中也可以包含所有消费者可能用得到的方法等,比如做些数据的加解密
packagecom.codecoord.rocketmq.domain;importlombok.Data;importjava.time.LocalDateTime;importjava.util.UUID;/**
 * 基础消息实体,包含基础的消息
 * 根据自己的业务消息设置更多的字段
 *
 * @author tianxincoord@163.com
 * @since 2022/6/16
 */@DatapublicabstractclassBaseMqMessage{/**
     * 业务键,用于RocketMQ控制台查看消费情况
     */protectedString key;/**
     * 发送消息来源,用于排查问题
     */protectedString source="";/**
     * 发送时间
     */protectedLocalDateTime sendTime=LocalDateTime.now();/**
     * 跟踪id,用于slf4j等日志记录跟踪id,方便查询业务链
     */protectedString traceId= UUID.randomUUID().toString();/**
     * 重试次数,用于判断重试次数,超过重试次数发送异常警告
     */protectedInteger retryTimes=0;}
  1. 有了此基础抽象实体类,那么剩下的所有业务消息实体只需要继承此基类,然后在自己业务类中包含自己需要的字段即可,因为这些公共字段不管是向上转型还是向下转型,子类和父类都可以看得到

2.2.2 RocketMQTemplate

  1. RocketMQTemplate发送消息的代码如果不封装,我们发送消息需要这样
    • String destination = topic + “:” + tag;
    • template.syncSend(destination, message);
  2. 每个人发送消息都要自己处理这个冒号,直接传入topic和tag不香吗?按照抽离变化点中的变化点,只有消息是变化的,除此之外的其他规则交给封装类
  3. RocketMQTemplate主要封装发送消息的日志、异常的处理、消息key设置、等等其他配置
  4. 封装代码类如下,下面包含了主要发送方式,更多自己添加即可
    • 这里就是消息发送的点餐机器,同时也提供了封装方法也提供原始RocketMQTemplate供使用
    • 此处只是提供一种方式,生产中按照项目组商量决定
packagecom.codecoord.rocketmq.template;importcom.alibaba.fastjson.JSONObject;importcom.codecoord.rocketmq.constant.RocketMqSysConstant;importcom.codecoord.rocketmq.domain.BaseMqMessage;importcom.codecoord.rocketmq.util.JsonUtil;importorg.apache.rocketmq.client.producer.SendResult;importorg.apache.rocketmq.spring.core.RocketMQTemplate;importorg.apache.rocketmq.spring.support.RocketMQHeaders;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importorg.springframework.messaging.Message;importorg.springframework.messaging.support.MessageBuilder;importorg.springframework.stereotype.Component;importjavax.annotation.Resource;/**
 * RocketMQ模板类
 *
 * @author tianxincoord@163.com
 * @since 2022/4/15
 */@ComponentpublicclassRocketMqTemplate{privatestaticfinalLogger LOGGER=LoggerFactory.getLogger(RocketMqTemplate.class);@Resource(name="rocketMQTemplate")privateRocketMQTemplate template;/**
     * 获取模板,如果封装的方法不够提供原生的使用方式
     */publicRocketMQTemplategetTemplate(){return template;}/**
     * 构建目的地
     */publicStringbuildDestination(String topic,String tag){return topic+RocketMqSysConstant.DELIMITER+ tag;}/**
     * 发送同步消息
     */public<TextendsBaseMqMessage>SendResultsend(String topic,String tag,T message){// 注意分隔符returnsend(topic+RocketMqSysConstant.DELIMITER+ tag, message);}public<TextendsBaseMqMessage>SendResultsend(String destination,T message){// 设置业务键,此处根据公共的参数进行处理// 更多的其它基础业务处理...Message<T> sendMessage=MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();SendResult sendResult= template.syncSend(destination, sendMessage);// 此处为了方便查看给日志转了json,根据选择选择日志记录方式,例如ELK采集
        LOGGER.info("[{}]同步消息[{}]发送结果[{}]", destination,JsonUtil.toJson(message),JSONObject.toJSON(sendResult));return sendResult;}/**
     * 发送延迟消息
     */public<TextendsBaseMqMessage>SendResultsend(String topic,String tag,T message,int delayLevel){returnsend(topic+RocketMqSysConstant.DELIMITER+ tag, message, delayLevel);}public<TextendsBaseMqMessage>SendResultsend(String destination,T message,int delayLevel){Message<T> sendMessage=MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();SendResult sendResult= template.syncSend(destination, sendMessage,3000, delayLevel);
        LOGGER.info("[{}]延迟等级[{}]消息[{}]发送结果[{}]", destination, delayLevel,JsonUtil.toJson(message),JsonUtil.toJson(sendResult));return sendResult;}}
  1. 这个类是最基础的原始封装类,相当于餐馆提供的点餐服务。上面提供无业务特性的发送,比如想要发送日志消息或者动态发送消息目的场景

3.2.3 增强RocketMQTemplate

  1. 以订单处理中心来说,变化点仅仅只是单号等业务数据不一样,发往订单处理中心的消息不管是topic还是tag等等其实完全都一样,那么此时可以根据业务来增加封装
  2. 增强原始功能需要注意下面两个点
    • 所有父类能出现的地方,子类都能出现:也就是子类拥有功能 >= 父类 ,比如Java的List,只要入参是List的地方,传ArrayList和LinkedList完全可以
    • 增强功能不能改变原始功能的行为:比如父类有一个方法say是说话,结果子类覆写了say改成了行为是吃饭,然后当调用者调用say的时候得到了一个完全预期外的结果
  3. 就以订单中心消息发送为例,封装OrderMessageTemplate继承自RocketMqTemplate,此时前者就拥有了封装父类的所有基础方法,拥有了所有父类的功能。然后可以在前者增加自身业务特性的发送方法,比如发送订单处理消息
packagecom.codecoord.rocketmq.template;importcom.codecoord.rocketmq.constant.RocketMqBizConstant;importcom.codecoord.rocketmq.domain.RocketMqEntityMessage;importorg.apache.rocketmq.client.producer.SendResult;importorg.springframework.stereotype.Component;importjavax.annotation.Resource;importjavax.validation.constraints.NotNull;importjava.time.LocalDate;importjava.time.LocalDateTime;/**
 * 订单类发送消息模板工具类
 *
 * @author tianxincode@163.com
 * @since 2022/6/16
 */@ComponentpublicclassOrderMessageTemplateextendsRocketMqTemplate{/// 如果不采用继承也可以直接注入使用/* @Resource
    private RocketMqTemplate rocketMqTemplate; *//**
     * 入参只需要传入是哪个订单号和业务体消息即可,其他操作根据需要处理
     * 这样对于调用者而言,可以更加简化调用
     */publicSendResultsendOrderPaid(@NotNullString orderId,String body){RocketMqEntityMessage message=newRocketMqEntityMessage();
        message.setKey(orderId);
        message.setSource("订单支付");
        message.setMessage(body);// 这两个字段只是为了测试
        message.setBirthday(LocalDate.now());
        message.setTradeTime(LocalDateTime.now());returnsend(RocketMqBizConstant.SOURCE_TOPIC,RocketMqBizConstant.ORDER_PAID_TAG, message);}}
  • 此时对于调用者只需要 orderMessageTemplate.sendOrderPaid(“O001”, “xxx”);就可以把消息发送到订单处理中心
  1. 封装后的好处,假如现在有10个订单来源,现在需要调整消息发送格式,如果不进行封装那么10个来源发送的地方都需要改;如果进行了二次封装,只需要改sendOrderPaid方法即可,而且还不会出错,此时优势就体现出来了

2.3 RocketMQListener封装

  1. RocketMQListener是消费消息的核心,同时也涉及到更多的操作,比如:基础日志记录、异常处理、消息重试、警告通知等等等
  2. 按照抽离变化点,RocketMQListener只应该处理与自身业务相关的,除此之外的其它应该交给父类,子类只需要告诉父类要不要异常处理、要不要重试等等,点餐式服务
  3. 封装消息消费的抽象类
    • 注意泛型限定为标准基础消息类,这样能到消费者的一定有统一的标准类BaseMqMessage
    • 下面简单封装示例
packagecom.codecoord.rocketmq.listener;importcom.codecoord.rocketmq.constant.RocketMqSysConstant;importcom.codecoord.rocketmq.domain.BaseMqMessage;importcom.codecoord.rocketmq.template.RocketMqTemplate;importcom.codecoord.rocketmq.util.JsonUtil;importorg.apache.rocketmq.client.producer.SendResult;importorg.apache.rocketmq.client.producer.SendStatus;importorg.apache.rocketmq.spring.annotation.RocketMQMessageListener;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importorg.slf4j.MDC;importjavax.annotation.Resource;importjava.time.Instant;importjava.util.Objects;/**
 * 抽象消息监听器,封装了所有公共处理业务,如
 * 1、基础日志记录
 * 2、异常处理
 * 3、消息重试
 * 4、警告通知
 * 5、....
 *
 * @author tianxincoord@163.com
 * @since 2022/4/17
 */publicabstractclassBaseMqMessageListener<TextendsBaseMqMessage>{/**
     * 这里的日志记录器是哪个子类的就会被哪个子类的类进行初始化
     */protectedfinalLogger logger=LoggerFactory.getLogger(this.getClass());@ResourceprivateRocketMqTemplate rocketMqTemplate;/**
     * 消息者名称
     *
     * @return 消费者名称
     */protectedabstractStringconsumerName();/**
     * 消息处理
     *
     * @param message 待处理消息
     * @throws Exception 消费异常
     */protectedabstractvoidhandleMessage(T message)throwsException;/**
     * 超过重试次数消息,需要启用isRetry
     *
     * @param message 待处理消息
     */protectedabstractvoidoverMaxRetryTimesMessage(T message);/**
     * 是否过滤消息,例如某些
     *
     * @param message 待处理消息
     * @return true: 本次消息被过滤,false:不过滤
     */protectedbooleanisFilter(T message){returnfalse;}/**
     * 是否异常时重复发送
     *
     * @return true: 消息重试,false:不重试
     */protectedabstractbooleanisRetry();/**
     * 消费异常时是否抛出异常
     *
     * @return true: 抛出异常,false:消费异常(如果没有开启重试则消息会被自动ack)
     */protectedabstractbooleanisThrowException();/**
     * 最大重试次数
     *
     * @return 最大重试次数,默认10次
     */protectedintmaxRetryTimes(){return10;}/**
     * isRetry开启时,重新入队延迟时间
     *
     * @return -1:立即入队重试
     */protectedintretryDelayLevel(){return-1;}/**
     * 由父类来完成基础的日志和调配,下面的只是提供一个思路
     */publicvoiddispatchMessage(T message){
        MDC.put(RocketMqSysConstant.TRACE_ID, message.getTraceId());// 基础日志记录被父类处理了
        logger.info("[{}]消费者收到消息[{}]",consumerName(),JsonUtil.toJson(message));if(isFilter(message)){
            logger.info("消息不满足消费条件,已过滤");return;}// 超过最大重试次数时调用子类方法处理if(message.getRetryTimes()>maxRetryTimes()){overMaxRetryTimesMessage(message);return;}try{long start=Instant.now().toEpochMilli();handleMessage(message);long end=Instant.now().toEpochMilli();
            logger.info("消息消费成功,耗时[{}ms]",(end- start));}catch(Exception e){
            logger.error("消息消费异常", e);// 是捕获异常还是抛出,由子类决定if(isThrowException()){thrownewRuntimeException(e);}if(isRetry()){// 获取子类RocketMQMessageListener注解拿到topic和tagRocketMQMessageListener annotation=this.getClass().getAnnotation(RocketMQMessageListener.class);if(Objects.nonNull(annotation)){
                    message.setSource(message.getSource()+"消息重试");
                    message.setRetryTimes</
  • 作者:TianXinCoord
  • 原文链接:https://blog.csdn.net/sinat_34104446/article/details/125355774
    更新时间:2022-07-17 14:47:48