spring cloud Alibaba 消息驱动RocketMQ学习笔记八

灰太狼 2022-12-20 02:28 165阅读 0赞

### 消息驱动RocketMQ ###

*  一、管理员审核分享业务实现
 *  二、RocketMQ安装、配置
 *  三、RocketMQ专业术语
 *  四、spring message（消息编程模型）
 *  五、spring cloud stream
 *   *  1、定义
     *  2、架构
     *  3、编程模型
     *  4、编写生产者
     *  5、编写消费者
     *  6、采用自定义接口，编写生产者
     *  7、采用自定义接口，编写消费者
     *  8、消息过滤
     *  9、监控-actuator
     *  10、异常处理
     *  11、Spring cloud steam + rocketMq实现分布式事务
     *  12、知识盘点

# 一、管理员审核分享业务实现 #

*  涉及逻辑：若分享不存在或分享处于审核状态，则抛异常；反之，进行资源审核，修改状态为PASS/REJUCT；PASS，为发布人添加积分

优化：若提升用户体验，此API主要流程是审核，为发布人添加积分为附属操作，并不需要等待返回，可以把加积分操作改造为异步执行，从而提升用户体验

实现异步方式：  
① AsyncRestTemplate  
② Async注解  
③ WebClient（Spring5.0引入，替代AsyncRestTemplate）  
④ MQ 消息队列  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 2]  
[常用MQ产品的对比][MQ]

# 二、RocketMQ安装、配置 #

[1、RocketMQ 4.5.1安装教程][1_RocketMQ 4.5.1]

[2、RocketMQ控制台安装教程][2_RocketMQ]

# 三、RocketMQ专业术语 #

[RocketMQ的概念与术语][RocketMQ]

# 四、spring message（消息编程模型） #

1、编写生产者

*  加依赖

<dependency>
       <groupId>org.apache.rocketmq</groupId>
       <artifactId>rocketmq-spring-boot-starter</artifactId>
       <version>2.1.1</version>
    </dependency>

*  加注解：无
 *  加配置

rocketmq:
      name-server: localhost:9876
      producer:
        #注意：需要指定group
        group: test-group

*  使用

private final RocketMQTemplate rocketMQTemplate;
    
    public Share auditById(Integer id, ShareAuditDto auditDto) { 
      //查询share是否存在，不存在或者当前audit_status != NOT_YET,那么就抛异常
        Share share = shareMapper.selectByPrimaryKey(id);
        if(share == null){ 
            throw new IllegalArgumentException("参数非法！该分享不存在！");
        }
        if(Objects.equals("NOT_YET",share.getAuditStatus())){ 
            throw new IllegalArgumentException("参数非法！该分享已审核！");
        }
        //审核资源，将状态设为pass/reject
        share.setAuditStatus(auditDto.getAuditStatusEnum().toString());
        shareMapper.updateByPrimaryKeySelective(share);
        //如果通过，发送消息给rocketMQ,让用户中心消费，并为发布人添加积分
        //采用异步执行
        rocketMQTemplate.convertAndSend(
            "add-bonus",
            UserAddBonusMsgDto.builder()
                .userId(share.getUserId())
                .bonus(50)
                .build()
        );
        return share;
    }

idea：下载RestFulTool，controller接口测试工具

小结:其他mq对应的template使用  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 3]  
2、编写消费者

package com.hzb2i.usercenter.rocketmq;
    
    import com.hzb2i.feignapi.domain.dto.message.UserAddBonusMsgDto;
    import com.hzb2i.usercenter.dao.bonusEventLog.BonusEventLogMapper;
    import com.hzb2i.usercenter.dao.user.UserMapper;
    import com.hzb2i.usercenter.domain.entity.bonusEventLog.BonusEventLog;
    import com.hzb2i.usercenter.domain.entity.user.User;
    import lombok.RequiredArgsConstructor;
    import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
    import org.apache.rocketmq.spring.core.RocketMQListener;
    import org.springframework.stereotype.Service;
    import java.util.Date;
    
    @RequiredArgsConstructor
    @Service
    @RocketMQMessageListener(consumerGroup = "consumer-group",topic = "add-bonus")
    public class AddBonusListener implements RocketMQListener<UserAddBonusMsgDto> { 
    
        private final UserMapper userMapper;
        private final BonusEventLogMapper bonusEventLogMapper;
    
        @Override
        public void onMessage(UserAddBonusMsgDto userAddBonusMsgDto) { 
            //当收到消息的时候，执行如下业务
            //1、为用户加积分
            User user = userMapper.selectByPrimaryKey(userAddBonusMsgDto.getUserId());
            user.setBonus(user.getBonus() + userAddBonusMsgDto.getBonus());
            userMapper.updateByPrimaryKeySelective(user);
            //2、记录日志到bonus_event_log表里
            bonusEventLogMapper.insert(BonusEventLog.builder()
                    .userid(userAddBonusMsgDto.getUserId())
                    .value(userAddBonusMsgDto.getBonus())
                    .createtime(new Date())
                    .event("contribute")
                    .description("投稿加积分")
                    .build());
        }
    }

小结：其他mq对应的listener  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 4]  
3、实现分布式事务  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 5]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 6]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 7]

# 五、spring cloud stream #

## 1、定义 ##

用于构建消息驱动的微服务框架，致力于简化mq通信的框架

## 2、架构 ##

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

## 3、编程模型 ##

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

*  destination binder：目标绑定器，与消息中间件通信的组件
 *  destination bindings：目标绑定，连接应用程序跟消息中间件的桥梁，用于消息的消费和生产，由binder创建
 *  message：消息

## 4、编写生产者 ##

*  加依赖

<dependency>
         <groupId>org.springframework.cloud</groupId>
         <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
     </dependency>

*  加注解：启动类加上 @EnableBinding(Source.class)
 *  写配置

spring:
    	cloud:
    	    stream:
    	      rocketmq:
    	        binder:
    	          name-server: 192.168.189.128:9876
    	      bindings:
    	        output:
    	          # 用来指定topic
    	          destination: stream-test-topic

*  代码示例

private final Source source;
    /** * spring cloud stream 发送消息 * @return */
    @GetMapping("test-stream")
    public String testStream(){ 
        source.output().send(MessageBuilder.withPayload("消息体").build());
        return "success";
    }

## 5、编写消费者 ##

*  加依赖
 *  加注解：启动类，加上@EnableBinding(Sink.class)
 *  写配置

## 6、采用自定义接口，编写生产者 ##

*  创建自定义接口：

public interface MySource { 
        String MY_OUTPUT = "my-output";
    
        @Output(MY_OUTPUT)
        MessageChannel output();
    }

*  加注解：启动类添加 @EnableBinding(\{Source.class, MySource.class\})
 *  写配置

spring:
    	 cloud:
    	    stream:
    	      rocketmq:
    	        binder:
    	          name-server: 192.168.189.128:9876
    	      bindings:
    	        output:
    	          # 用来指定topic
    	          destination: stream-test-topic
    	        #要和定义的接口名称一致
    	        my-output:
    	          # 用来指定topic
    	          destination: stream-my-topic

*  代码示例

private final MySource mySource;
    
    /** * spring cloud stream 发送消息 * @return */
    @GetMapping("test-my-stream")
    public String testMyStream(){ 
        mySource.output().send(MessageBuilder.withPayload("自定义消息体").build());
        return "success";
    }

小坑：mybatis报错，若MapperScan扫描的范围过大，会将自定义的stream接口也当做是mapper接口，而此接口又没有对应的xml文件，就会报错，解决方式，缩小MapperScan扫描范围即可

## 7、采用自定义接口，编写消费者 ##

*  创建自定义接口

@Service
    public interface MySink { 
        String MY_INPUT = "my_input";
    
        @Input(MY_INPUT)
        SubscribableChannel input();
    }

*  加注解：启动类，添加@EnableBinding(\{Sink.class, MySink.class\})
 *  写配置

spring:
    	cloud:
    	   stream:
    	     rocketmq:
    	       binder:
    	         name-server: 192.168.189.128:9876
    	     bindings:
    	       input:
    	         # 用来指定topic
    	         destination: stream-test-topic
    	         # rocketMQ一定要设置，不然无法启动，其他MQ可以不设置
    	         group: binder-group
    	       my_input:
    	         # 用来指定topic
    	         destination: stream-my-topic
    	         # rocketMQ一定要设置，不然无法启动，其他MQ可以不设置
    	         group: my-binder-group

*  代码示例：

@StreamListener(MySink.MY_INPUT)
    public void MyReceive(String messageBody){ 
        log.info("通过stream，结合自定义接口收到消息：messageBody={}",messageBody);
    }

mybatis小坑，解决方式同上

小结：

*  source（发送接口）、sink（接受接口）、processor（发送、接受接口）
 *  现象（会用）、本质（理解，看源码，多总结）：配置类上的名称为什么一定要和接口定义的名称一致？因为spring有IOC，会通过名称注入并创建对象

## 8、消息过滤 ##

[Spring Cloud Stream实现消息过滤消费][Spring Cloud Stream]

## 9、监控-actuator ##

*  首页：http://localhost:8088/actuator
 *  bindings(应用程序-消息中间件桥梁)运行信息：http://localhost:8088/actuator/bindings
 *  channels（通道）：http://localhost:8088/actuator/channels
 *  health（应用运行状态）：http://localhost:8088/actuator/health  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 10]

## 10、异常处理 ##

[Spring Cloud Stream错误处理详解][Spring Cloud Stream 1]

## 11、Spring cloud steam + rocketMq实现分布式事务 ##

## 12、知识盘点 ##

[Spring Cloud Stream知识点盘点][Spring Cloud Stream 2]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/045edd8f0e3348e0bd6e9b3faf67b9f7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/ed9df145deb24940b50dd8d5261c362a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/cf59e07c11f6413e8ddd59788943dc52.png
[MQ]: https://www.imooc.com/article/290040
[1_RocketMQ 4.5.1]: https://www.imooc.com/article/290089
[2_RocketMQ]: https://www.imooc.com/article/290092
[RocketMQ]: https://www.jianshu.com/p/3938d990a8de
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221120/07834ea94379494b8ea0edee8b6db3f6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 4]: https://img-blog.csdnimg.cn/20201114215108945.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY=,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 5]: https://img-blog.csdnimg.cn/20201114220223158.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY=,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 6]: https://img-blog.csdnimg.cn/20201114220643765.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY=,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 7]: https://img-blog.csdnimg.cn/20201114220714509.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY=,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 8]: https://img-blog.csdnimg.cn/20201115204352257.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY=,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 9]: https://img-blog.csdnimg.cn/20201115204531917.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY=,size_16,color_FFFFFF,t_70#pic_center
[Spring Cloud Stream]: https://www.imooc.com/article/290424
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY_size_16_color_FFFFFF_t_70_pic_center 10]: https://img-blog.csdnimg.cn/20201119222917795.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6ZjIyODQ0NjY=,size_16,color_FFFFFF,t_70#pic_center
[Spring Cloud Stream 1]: https://www.imooc.com/article/290435
[Spring Cloud Stream 2]: https://www.imooc.com/article/290489