【SpringBoot框架篇】22.集成RocketMQ实现消息发布和订阅

- 日理万妓 2022-08-29 00:23 291阅读 0赞

### 文章目录 ###

*  一.相关文章
 *  二.使用的依赖
 *  三.配置文件
 *  四.发送正常的消息
 *   *  1.消费者（订阅者）
     *  2.发布者(生产者)
     *  3.测试
 *  五.发布事务消息结合@Transactional一起使用
 *   *  1.消费者（订阅者）
     *  2.生产者 (发布者)
     *  3.测试
     *   *  3.1 正常测试
         *  3.2 回滚mysql事务测试
 *  六.项目配套代码

# 一.相关文章 #

RocketMQ简介请看这篇文章: [RocketMQ入门笔记][RocketMQ]  
RocketMQ安装请看这篇文章: [使用docker安装RocketMQ][docker_RocketMQ]

# 二.使用的依赖 #

<dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-spring-boot-starter</artifactId>
                <version>2.1.0</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-data-jpa</artifactId>
            </dependency>
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <version>8.0.11</version>
            </dependency>

# 三.配置文件 #

rocketmq:
      #namesrv访问地址
      name-server: 127.0.0.1:9876
      producer:
        #消息生产者组名
        group: test-group
        #发送消息的超时时间
        send-message-timeout: 3000
        #异步消息重试的次数
        retry-times-when-send-async-failed: 2
        #重试是否换一个server
        retry-next-server: true
        #同步消息重试的次数
        retry-times-when-send-failed: 2
        #消费消息认证用的ak
        access-key: Ak
        #消费消息认证用的sk
        secret-key: SK
        
    topic:
      string: stringTopic
      order: orderTopic
      
    spring:
      application:
        name: springboot-rocketMq
      datasource:
        driver-class-name: com.mysql.cj.jdbc.Driver
        url: jdbc:mysql://localhost:3306/test?createDatabaseIfNotExist=true&useSSL=false&serverTimezone=GMT%2b8&characterEncoding=utf8&connectTimeout=10000&socketTimeout=3000&autoReconnect=true
        username: root
        password: 123456
      jpa:
        database: mysql
        show-sql: true
        hibernate:
          #自动创建或修改表结构
          ddl-auto: update
          naming:
            physical-strategy: org.hibernate.boot.model.naming.PhysicalNamingStrategyStandardImpl

# 四.发送正常的消息 #

## 1.消费者（订阅者） ##

注意事项

*  消息发布者设置**topic** 要和消费者的**RocketMQMessageListener注解**里的**topic** 一致。
 *  RocketMQListener泛型类型要和生产者发送的消息的类型一致,如果生产者发布的类型是User，此处的String要改成User.

/** * @author Dominick Li * @CreateTime 2020/3/22 21:06 * @description 如果consumerGroup的名称不同, 则会产生重复消费的情况，例如 consumer1和consumer2 * 如果一个top下面有多个consumerGroup组,则消费者 消费消息会根据组的长度取模 **/
    @Service
    @RocketMQMessageListener(topic = "${topic.string}", consumerGroup = "${spring.application.name}-${topic.string}-consumer1")
    public class StringConsumer implements RocketMQListener<String> { 
        @Override
        public void onMessage(String msg) { 
            System.out.println("StringConsumer1开始消费:"+ msg);
    // if(true){ 
    // //模拟出现异常,出现异常后这个任务还会被重试消费
    // System.out.println(1/0);
    // }
        }
    }

## 2.发布者(生产者) ##

发送消息的三种方式

<table> 
 <thead> 
  <tr> 
   <th>方法</th> 
   <th>描述</th> 
   <th>请求类型</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>void send(topic,msg)</td> 
   <td>底层调用的syncSend方法，但是不接受返回结果</td> 
   <td>异步</td> 
  </tr> 
  <tr> 
   <td>SendResult syncSend(topic,msg)</td> 
   <td>发送消息,等待发送消息返回的结果</td> 
   <td>同步</td> 
  </tr> 
  <tr> 
   <td>void asyncSend(topic,msg,callback)</td> 
   <td>发送消息,通过回调函数根据发送状态处理相对于逻辑</td> 
   <td>异步</td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>topic</td> 
   <td>消息队列</td> 
  </tr> 
  <tr> 
   <td>msg</td> 
   <td>消息类型</td> 
  </tr> 
  <tr> 
   <td>callback</td> 
   <td>回调函数</td> 
  </tr> 
  <tr> 
   <td>SendResult</td> 
   <td>发送消息是否成功,SendResult.OK为发送成功,其它为失败。</td> 
  </tr> 
 </tbody> 
</table>

@Resource
        private RocketMQTemplate rocketMQTemplate;
        @Value("${topic.string}")
        private String stringTopic;
        @Test
        void contextLoads() { 
            testBaseMethod();
        }
        public void testBaseMethod() { 
            //1.发送异步消息,但是不会确认消息有没有被接收,日志可以这么发,如果是对数据一致性要去比较高的,建议使用下面的方法
            rocketMQTemplate.send(stringTopic, MessageBuilder.withPayload(String.format("test send method")).build());
            //2. 发送同步消息,等待发送消息返回的结果
            SendResult sendResult = rocketMQTemplate.syncSend(stringTopic, "test syncSend method");
            if (sendResult.getSendStatus() == SendStatus.SEND_OK) { 
                System.out.println("发送成功...");
            }
            //3.发送异步消息,回调里处理发送成功或失败逻辑
            for (int j = 1; j < 3; j++) { 
                rocketMQTemplate.asyncSend(stringTopic, "test asyncSend method", new SendCallback() { 
                    @Override
                    public void onSuccess(SendResult sr) { 
                        if (sr.getSendStatus() == SendStatus.SEND_OK) { 
                            System.out.print("async onSucess ok");
                        } else { 
                            System.out.print("async onSucess fail");
                        }
                    }
                    @Override
                    public void onException(Throwable var1) { 
                        System.out.printf("async onException Throwable=%s %n", var1);
                    }
                });
                System.out.printf("发送第%d条消息\n", j);
            }
        }

## 3.测试 ##

执行发布消息的函数，然后控制台输出如下  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70]

# 五.发布事务消息结合@Transactional一起使用 #

rocketmq事务消息是发生在Producer和Broker之间，是二阶段提交。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70 1]  
第一阶段是：步骤1，2，3。  
第二阶段是：步骤4，5。

## 1.消费者（订阅者） ##

和上面定义的消费者的除了泛型和消费队列不一样,其它基本是一致的。

@Service
    @RocketMQMessageListener(topic = "${topic.order}", consumerGroup = "${spring.application.name}-${topic.order}-consumer1")
    public class OrderConsumer implements RocketMQListener<Orders> { 
        @Override
        public void onMessage(Orders orders) { 
            System.out.println("OrderConsumer1开始消费" + orders.getName() + "," + orders.getOrderId());
        }
    }

## 2.生产者 (发布者) ##

**执行逻辑描述**

*  1.发送事务消息后**TransactionListener** 会监听到事务消息,并执行MQ本地事务executeLocalTransaction方法
 *  2.在**executeLocalTransaction**方法中执行添加数据到数据库的操作,被执行的方法需要添加\*\*@Transactional注解\*\*让数据库事务可以回滚。
 *  3.在**checkLocalTransaction**方法中会检查本地事务是否提交成功,根据消息生产者传递的事务Id(订单Id)去查询数据库里面是否存在这条记录,如果记录存在提交MQ事务,如果不存在则表示数据库事务没有正常提交,MQ事务也进行回滚。

发送端代码

/** * 发送事务消息 */
        public void testTransactionMethod() { 
            Long orderId = System.currentTimeMillis();
            Orders order = new Orders();
            order.setOrderId(orderId);
            order.setName("酒水订单");
            order.setCreateDate(new Date());
            rocketMQTemplate.sendMessageInTransaction(orderTopic,
                    MessageBuilder.withPayload(
                            order)
                            .setHeader(RocketMQHeaders.TRANSACTION_ID, orderId)
                            .build()
                    , order);
        }

执行MQ事务代码

@RocketMQTransactionListener
    public class TransactionListener implements RocketMQLocalTransactionListener { 
    
        @Autowired
        OrdersService ordersService;
    
        /** * 执行本地事务 */
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) { 
            String transId = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
            System.out.printf("#### executeLocalTransaction transactionId=%s %n",
                    transId);
            //执行本地事务，并记录日志
            ordersService.save((Orders) arg);
            //执行成功，可以提交事务
            return RocketMQLocalTransactionState.COMMIT;
    
        }
    
        /** * 检查本地事务是否成功 */
        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message msg) { 
            String transId = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
            System.out.printf("#### checkLocalTransaction transactionId=%s %n", transId);
            Optional<Orders> ordersOptional = ordersService.findById(Long.parseLong(transId));
            if (ordersOptional.isPresent()) { 
                return RocketMQLocalTransactionState.COMMIT;
            } else { 
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        }
    
    }

数据库操作相关代码

@Component
    public class OrdersService { 
    
        @Autowired
        OrdersRepository ordersRepository;
    
        @Transactional
        public Orders save(Orders orders) { 
            Orders save = ordersRepository.save(orders);
            if (false) { 
                //测试抛出RuntimeException让mysql事务回滚
                System.out.println(1 / 0);
            }
            return save;
        }
        
        public Optional<Orders> findById(Long id) { 
            return ordersRepository.findById(id);
        }
    
    }
    
    @Entity
    @Table(name = "orders")
    public class Orders implements Serializable { 
        @Id
        private Long orderId;
        private String name;
        private Date createDate; 
        //省略get,set方法
    }
    public interface OrdersRepository extends JpaRepository<Orders,Long> { 
    }

## 3.测试 ##

### 3.1 正常测试 ###

此操作不模拟异常,正常发送事务消息,可以看到插入数据库的事务提交和MQ的事务提交都同步提交了。  
![在这里插入图片描述][37c1712d3e8047e2a52dd2092ee32d15.png]![在这里插入图片描述][037d6c9cd58048bfbe6986b0d0af23cd.png]  
![在这里插入图片描述][fd7551fa806e4ba3af7d2bd89d5650ed.png]

### 3.2 回滚mysql事务测试 ###

在执行数据库操作处把if(false)改成if(true)，模拟数据库异常让事务回滚，然后发送事务消息。  
![在这里插入图片描述][4c4f8fb40f3148ec88a3804f2df725ca.png]  
执行后的结果如下,可以看到控制台并没有打印消费消息的信息  
![在这里插入图片描述][c902852b545b40bbb06075c3b7913c90.png]  
查看数据库也可以发现没有orderId为1628439222274的记录。  
![在这里插入图片描述][ba2b45ebc2e0412dac573d022e533502.png]

# 六.项目配套代码 #

[github地址][github]

创作不易,要是觉得我写的对你有点帮助的话,麻烦在github上帮我点下 **Star**

【SpringBoot框架篇】其它文章如下,后续会继续更新。

*  [1.搭建第一个springboot项目][1._springboot]
 *  [2.Thymeleaf模板引擎实战][2.Thymeleaf]
 *  [3.优化代码,让代码更简洁高效][3.]
 *  [4.集成jta-atomikos实现分布式事务][4._jta-atomikos]
 *  [5.分布式锁的实现方式][5.]
 *  [6.docker部署,并挂载配置文件到宿主机上面][6.docker]
 *  [7.项目发布到生产环境][7.]
 *  [8.搭建自己的spring-boot-starter][8._spring-boot-starter]
 *  [9.dubbo入门实战][9.dubbo]
 *  [10.API接口限流实战][10.API]
 *  [11.Spring Data Jpa实战][11.Spring Data Jpa]
 *  [12.使用druid的monitor工具查看sql执行性能][12._druid_monitor_sql]
 *  [13.使用springboot admin对springboot应用进行监控][13._springboot admin_springboot]
 *  [14.mybatis-plus实战][14.mybatis-plus]
 *  [15.使用shiro对web应用进行权限认证][15._shiro_web]
 *  [16.security整合jwt实现对前后端分离的项目进行权限认证][16.security_jwt]
 *  [17.使用swagger2生成RESTful风格的接口文档][17._swagger2_RESTful]
 *  [18.使用Netty加websocket实现在线聊天功能][18._Netty_websocket]
 *  [19.使用spring-session加redis来实现session共享][19._spring-session_redis_session]
 *  [20.自定义@Configuration配置类启用开关][20._Configuration]
 *  [21.对springboot框架编译后的jar文件瘦身][21._springboot_jar]
 *  [22.集成RocketMQ实现消息发布和订阅][22._RocketMQ]

[RocketMQ]: https://dominick-li.blog.csdn.net/article/details/117458872
[docker_RocketMQ]: https://dominick-li.blog.csdn.net/article/details/109063041
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70]: /images/20220829/7fe2e37de93d4deaafd435d9da3144c0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70 1]: /images/20220829/2c96f03c39794e4e85d800871d4f8dda.png
[37c1712d3e8047e2a52dd2092ee32d15.png]: https://img-blog.csdnimg.cn/37c1712d3e8047e2a52dd2092ee32d15.png
[037d6c9cd58048bfbe6986b0d0af23cd.png]: /images/20220829/b8c6bb533d584699abc56c48ae2005ed.png
[fd7551fa806e4ba3af7d2bd89d5650ed.png]: https://img-blog.csdnimg.cn/fd7551fa806e4ba3af7d2bd89d5650ed.png
[4c4f8fb40f3148ec88a3804f2df725ca.png]: https://img-blog.csdnimg.cn/4c4f8fb40f3148ec88a3804f2df725ca.png
[c902852b545b40bbb06075c3b7913c90.png]: https://img-blog.csdnimg.cn/c902852b545b40bbb06075c3b7913c90.png
[ba2b45ebc2e0412dac573d022e533502.png]: https://img-blog.csdnimg.cn/ba2b45ebc2e0412dac573d022e533502.png
[github]: https://github.com/Dominick-Li/springboot-master
[1._springboot]: https://blog.csdn.net/ming19951224/article/details/106159309
[2.Thymeleaf]: https://blog.csdn.net/ming19951224/article/details/106160322
[3.]: https://blog.csdn.net/ming19951224/article/details/106173840
[4._jta-atomikos]: https://blog.csdn.net/ming19951224/article/details/105753333
[5.]: https://blog.csdn.net/ming19951224/article/details/106205332
[6.docker]: https://blog.csdn.net/ming19951224/article/details/105923548
[7.]: https://blog.csdn.net/ming19951224/article/details/106212271
[8._spring-boot-starter]: https://blog.csdn.net/ming19951224/article/details/106224172
[9.dubbo]: https://blog.csdn.net/ming19951224/article/details/106225928
[10.API]: https://blog.csdn.net/ming19951224/article/details/106227451
[11.Spring Data Jpa]: https://blog.csdn.net/ming19951224/article/details/106308544
[12._druid_monitor_sql]: https://blog.csdn.net/ming19951224/article/details/106317893
[13._springboot admin_springboot]: https://blog.csdn.net/ming19951224/article/details/107138845
[14.mybatis-plus]: https://blog.csdn.net/ming19951224/article/details/107145780
[15._shiro_web]: https://blog.csdn.net/ming19951224/article/details/107192585
[16.security_jwt]: https://blog.csdn.net/ming19951224/article/details/107733689
[17._swagger2_RESTful]: https://blog.csdn.net/ming19951224/article/details/107741541
[18._Netty_websocket]: https://blog.csdn.net/ming19951224/article/details/108555917
[19._spring-session_redis_session]: https://blog.csdn.net/ming19951224/article/details/112758560
[20._Configuration]: https://blog.csdn.net/ming19951224/article/details/112759717
[21._springboot_jar]: https://dominick-li.blog.csdn.net/article/details/119024251
[22._RocketMQ]: https://dominick-li.blog.csdn.net/article/details/119523197