devbox
盐析白兔
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Spring Boot+RocketMQ 实现多实例分布式环境下的事件驱动_springboot应用内部用事件监听还是mq

作者：盐析白兔 | 2024-05-14 06:25:47

赞

踩

springboot应用内部用事件监听还是mq

技术老男孩 2024-01-07 16:00 发表于广东

技术老男孩

分享技术路上的点滴，专注于后端技术，助力开发者成长，欢迎关注。

55篇原创内容

公众号

1、为什么要使用MQ？

在Spring Boot Event这篇文章中已经通过Guava或者SpringBoot自身的Listener实现了事件驱动，已经做到了对业务的解耦。为什么还要用到MQ来进行业务解耦呢？

首先无论是通过Guava还是Spring Boot自身提供的监听注解来实现的事件驱动他都是处于同一进程中的，意思就是当前事件推送后只有当前的进程可以进行消费。

通过MQ可以实现将事件推送到进程外的Broker中，在多实例/分布式环境下，其他的服务在订阅同一事件（Topic）时，可以在各自的服务中进行消费，最大化空闲服务的利用。

图片

源码地址：

https://gitee.com/sparkle3021/springboot3-study

2、整合RocketMQ

依赖版本

  • JDK 17

  • Spring Boot 3.2.0

  • RocketMQ-Client 5.0.4

  • RocketMQ-Starter 2.2.0

Spring Boot 3.0+ 取消了对spring.factories的支持。所以在导入时需要手动引入RocketMQ的配置类。

引入RocketMQ依赖
  1. <dependency>
  2.     <groupId>org.apache.rocketmq</groupId>
  3.     <artifactId>rocketmq-client-java</artifactId>
  4.     <version>5.0.4</version>
  5. </dependency>
  6. <dependency>
  7.     <groupId>org.apache.rocketmq</groupId>
  8.     <artifactId>rocketmq-spring-boot-starter</artifactId>
  9.     <version>2.2.0</version>
  10. </dependency>
解决Spring Boot3+不兼容 spring.factories

rocketmq-spring-boot-starter:2.2.2版本中：

图片

参考配置文件
  1. # RocketMQ 配置
  2. rocketmq:
  3.   name-server: 127.0.0.1:9876
  4.   consumer:
  5.     group: event-mq-group
  6.     # 一次拉取消息最大值，注意是拉取消息的最大值而非消费最大值
  7.     pull-batch-size: 1
  8.   producer:
  9.     # 发送同一类消息的设置为同一个group，保证唯一
  10.     group: event-mq-group
  11.     # 发送消息超时时间，默认3000
  12.     sendMessageTimeout: 10000
  13.     # 发送消息失败重试次数，默认2
  14.     retryTimesWhenSendFailed: 2
  15.     # 异步消息重试此处，默认2
  16.     retryTimesWhenSendAsyncFailed: 2
  17.     # 消息最大长度，默认1024 * 1024 * 4(默认4M)
  18.     maxMessageSize: 4096
  19.     # 压缩消息阈值，默认4k(1024 * 4)
  20.     compressMessageBodyThreshold: 4096
  21.     # 是否在内部发送失败时重试另一个broker，默认false
  22.     retryNextServer: false

参考Issue

  • 方法一 ：通过@Import(RocketMQAutoConfiguration.class)在配置类中引入

  • 方法二：在resources资源目录下创建文件夹及文件META-INF/spring，org.springframework.boot.autoconfigure.AutoConfiguration.imports。

文件内容为RocketMQ自动配置类路径：org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration

3、RocketMQ 使用

解决Spring Boot3+不支持spring.factories的问题
  1. import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration;
  2. import org.springframework.boot.SpringApplication;
  3. import org.springframework.boot.autoconfigure.SpringBootApplication;
  4. import org.springframework.context.annotation.Import;
  5.  
  6. /**
  7.  * 启动类
  8.  */
  9. @Import(RocketMQAutoConfiguration.class)
  10. @SpringBootApplication
  11. public class MQEventApplication {
  12.     public static void main(String[] args) {
  13.         SpringApplication.run(MQEventApplication.class, args);
  14.     }
  15. }
RocketMQ操作工具

RocketMQ Message实体

  1. import cn.hutool.core.util.IdUtil;
  2. import jakarta.validation.constraints.NotBlank;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Builder;
  5. import lombok.Data;
  6. import lombok.NoArgsConstructor;
  7. import org.apache.commons.collections.CollectionUtils;
  8. import org.apache.commons.lang3.ObjectUtils;
  9. import org.springframework.messaging.Message;
  10. import org.springframework.messaging.support.MessageBuilder;
  11.  
  12. import java.io.Serializable;
  13. import java.util.List;
  14.  
  15. /**
  16.  * RocketMQ 消息
  17.  */
  18. @Data
  19. @Builder
  20. @AllArgsConstructor
  21. @NoArgsConstructor
  22. public class RocketMQMessage<T> implements Serializable {
  23.  
  24.     /**
  25.      * 消息队列主题
  26.      */
  27.     @NotBlank(message = "MQ Topic 不能为空")
  28.     private String topic;
  29.  
  30.     /**
  31.      * 延迟级别
  32.      */
  33.     @Builder.Default
  34.     private DelayLevel delayLevel = DelayLevel.OFF;
  35.  
  36.     /**
  37.      * 消息体
  38.      */
  39.     private T message;
  40.  
  41.     /**
  42.      * 消息体
  43.      */
  44.     private List<T> messages;
  45.  
  46.     /**
  47.      * 使用有序消息发送时，指定发送到队列
  48.      */
  49.     private String hashKey;
  50.  
  51.     /**
  52.      * 任务Id，用于日志打印相关信息
  53.      */
  54.     @Builder.Default
  55.     private String taskId = IdUtil.fastSimpleUUID();
  56. }

RocketMQTemplate 二次封装

  1. import com.yiyan.study.domain.RocketMQMessage;
  2. import jakarta.annotation.Resource;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.commons.lang3.StringUtils;
  5. import org.apache.rocketmq.client.producer.SendCallback;
  6. import org.apache.rocketmq.client.producer.SendResult;
  7. import org.apache.rocketmq.spring.core.RocketMQTemplate;
  8. import org.springframework.beans.factory.annotation.Value;
  9. import org.springframework.stereotype.Component;
  10.  
  11. /**
  12.  * RocketMQ 消息工具类
  13.  */
  14. @Slf4j
  15. @Component
  16. public class RocketMQService {
  17.  
  18.     @Resource
  19.     private RocketMQTemplate rocketMQTemplate;
  20.  
  21.     @Value("${rocketmq.producer.sendMessageTimeout}")
  22.     private int sendMessageTimeout;
  23.  
  24.     /**
  25.      * 异步发送消息回调
  26.      *
  27.      * @param taskId 任务Id
  28.      * @param topic  消息主题
  29.      * @return the send callback
  30.      */
  31.     private static SendCallback asyncSendCallback(String taskId, String topic) {
  32.         return new SendCallback() {
  33.             @Override
  34.             public void onSuccess(SendResult sendResult) {
  35.                 log.info("ROCKETMQ 异步消息发送成功 : [TaskId:{}] - [Topic:{}] - [SendStatus:{}]", taskId, topic, sendResult.getSendStatus());
  36.             }
  37.  
  38.             @Override
  39.             public void onException(Throwable throwable) {
  40.                 log.error("ROCKETMQ 异步消息发送失败 : [TaskId:{}] - [Topic:{}] - [ErrorMessage:{}]", taskId, topic, throwable.getMessage());
  41.             }
  42.         };
  43.     }
  44.  
  45.     /**
  46.      * 发送同步消息，使用有序发送请设置HashKey
  47.      *
  48.      * @param message 消息参数
  49.      */
  50.     public <T> void syncSend(RocketMQMessage<T> message) {
  51.         log.info("ROCKETMQ 同步消息发送 : [TaskId:{}] - [Topic:{}]", message.getTaskId(), message.getTopic());
  52.         SendResult sendResult;
  53.         if (StringUtils.isNotBlank(message.getHashKey())) {
  54.             sendResult = rocketMQTemplate.syncSendOrderly(message.getTopic(), message.getMessage(), message.getHashKey());
  55.         } else {
  56.             sendResult = rocketMQTemplate.syncSend(message.getTopic(), message.getMessage(), sendMessageTimeout, message.getDelayLevel().getLevel());
  57.         }
  58.         log.info("ROCKETMQ 同步消息发送结果 : [TaskId:{}] - [Topic:{}] - [MessageId:{}] - [SendStatus:{}]",
  59.                 message.getTaskId(), message.getTopic(), sendResult.getMsgId(), sendResult.getSendStatus());
  60.     }
  61.  
  62.     /**
  63.      * 批量发送同步消息
  64.      *
  65.      * @param message 消息参数
  66.      */
  67.     public <T> void syncSendBatch(RocketMQMessage<T> message) {
  68.         log.info("ROCKETMQ 同步消息-批量发送 : [TaskId:{}] - [Topic:{}] - [MessageCount:{}]",
  69.                 message.getTaskId(), message.getTopic(), message.getMessages().size());
  70.         SendResult sendResult;
  71.         if (StringUtils.isNotBlank(message.getHashKey())) {
  72.             sendResult = rocketMQTemplate.syncSendOrderly(message.getTopic(), message.getMessages(), message.getHashKey());
  73.         } else {
  74.             sendResult = rocketMQTemplate.syncSend(message.getTopic(), message.getMessages());
  75.         }
  76.         log.info("ROCKETMQ 同步消息-批量发送结果 : [TaskId:{}] - [Topic:{}] - [MessageId:{}] - [SendStatus:{}]",
  77.                 message.getTaskId(), message.getTopic(), sendResult.getMsgId(), sendResult.getSendStatus());
  78.     }
  79.  
  80.     /**
  81.      * 异步发送消息，异步返回消息结果
  82.      *
  83.      * @param message 消息参数
  84.      */
  85.     public <T> void asyncSend(RocketMQMessage<T> message) {
  86.         log.info("ROCKETMQ 异步消息发送 : [TaskId:{}] - [Topic:{}]", message.getTaskId(), message.getTopic());
  87.         if (StringUtils.isNotBlank(message.getHashKey())) {
  88.             rocketMQTemplate.asyncSendOrderly(message.getTopic(), message.getMessage(), message.getHashKey(),
  89.                     asyncSendCallback(message.getTaskId(), message.getTopic()));
  90.         } else {
  91.             rocketMQTemplate.asyncSend(message.getTopic(), message.getMessage(),
  92.                     asyncSendCallback(message.getTaskId(), message.getTopic()), sendMessageTimeout, message.getDelayLevel().getLevel());
  93.         }
  94.     }
  95.  
  96.     /**
  97.      * 批量异步发送消息
  98.      *
  99.      * @param message 消息参数
  100.      */
  101.     public <T> void asyncSendBatch(RocketMQMessage<T> message) {
  102.         log.info("ROCKETMQ 异步消息-批量发送 : [TaskId:{}] - [Topic:{}] - [MessageCount:{}]",
  103.                 message.getTaskId(), message.getTopic(), message.getMessages().size());
  104.         if (StringUtils.isNotBlank(message.getHashKey())) {
  105.             rocketMQTemplate.asyncSendOrderly(message.getTopic(), message.getMessages(), message.getHashKey(),
  106.                     asyncSendCallback(message.getTaskId(), message.getTopic()));
  107.         } else {
  108.             rocketMQTemplate.asyncSend(message.getTopic(), message.getMessages(),
  109.                     asyncSendCallback(message.getTaskId(), message.getTopic()));
  110.         }
  111.     }
  112.  
  113.     /**
  114.      * 单向发送消息，不关心返回结果，容易消息丢失，适合日志收集、不精确统计等消息发送;
  115.      *
  116.      * @param message 消息参数
  117.      */
  118.     public <T> void sendOneWay(RocketMQMessage<T> message) {
  119.         sendOneWay(message, false);
  120.     }
  121.  
  122.     /**
  123.      * 单向消息 - 批量发送
  124.      *
  125.      * @param message 消息体
  126.      * @param batch   是否为批量操作
  127.      */
  128.     public <T> void sendOneWay(RocketMQMessage<T> message, boolean batch) {
  129.         log.info((batch ? "ROCKETMQ 单向消息发送 : [TaskId:{}] - [Topic:{}]"
  130.                         : "ROCKETMQ 单向消息-批量发送 : [TaskId:{}] - [Topic:{}] - [MessageCount{}]"),
  131.                 message.getTaskId(), message.getTopic(), message.getMessages().size());
  132.         if (StringUtils.isNotBlank(message.getHashKey())) {
  133.             if (batch) {
  134.                 message.getMessages().
  135.                         forEach(msg -> rocketMQTemplate.sendOneWayOrderly(message.getTopic(), msg, message.getHashKey()));
  136.             } else {
  137.                 rocketMQTemplate.sendOneWayOrderly(message.getTopic(), message.getMessage(), message.getHashKey());
  138.             }
  139.         } else {
  140.             if (batch) {
  141.                 message.getMessages().forEach(msg -> rocketMQTemplate.sendOneWay(message.getTopic(), msg));
  142.             } else {
  143.                 rocketMQTemplate.sendOneWay(message.getTopic(), message.getMessage());
  144.             }
  145.         }
  146.     }
  147. }
定义RocketMQ消费者
  1. import com.yiyan.study.constants.MQConfig;
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
  4. import org.apache.rocketmq.spring.core.RocketMQListener;
  5. import org.springframework.stereotype.Component;
  6.  
  7. /**
  8.  * MQ消息监听
  9.  */
  10. @Component
  11. @Slf4j
  12. @RocketMQMessageListener(topic = MQConfig.EVENT_TOPIC,
  13.         consumerGroup = MQConfig.EVENT_CONSUMER_GROUP)
  14. public class MQListener implements RocketMQListener<String> {
  15.     @Override
  16.     public void onMessage(String message) {
  17.         log.info("MQListener 接收消息 ： {}", message);
  18.     }
  19. }
定义测试类发送消息
  1. import cn.hutool.core.thread.ThreadUtil;
  2. import com.yiyan.study.constants.MQConfig;
  3. import com.yiyan.study.domain.RocketMQMessage;
  4. import com.yiyan.study.utils.RocketMQService;
  5. import jakarta.annotation.Resource;
  6. import org.junit.jupiter.api.Test;
  7. import org.springframework.boot.test.context.SpringBootTest;
  8.  
  9. /**
  10.  * MQ测试
  11.  */
  12. @SpringBootTest
  13. public class MQTest {
  14.  
  15.     @Resource
  16.     private RocketMQService rocketMQService;
  17.  
  18.     @Test
  19.     public void sendMessage() {
  20.         int count = 1;
  21.         while (count <= 50) {
  22.             rocketMQService.syncSend(RocketMQMessage.builder()
  23.                     .topic(MQConfig.EVENT_TOPIC)
  24.                     .message(count++)
  25.                     .build());
  26.         }
  27.         // 休眠等待消费消息
  28.         ThreadUtil.sleep(2000L);
  29.     }
  30. }

4、测试

图片

原文：

blog.csdn.net/m0_55712478/

article/details/135242345

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/盐析白兔/article/detail/567714
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号