diff --git a/src/main/java/cn/iocoder/dashboard/framework/redis/config/RedisConfig.java b/src/main/java/cn/iocoder/dashboard/framework/redis/config/RedisConfig.java index 3c88ce4b2..13cda9dbb 100644 --- a/src/main/java/cn/iocoder/dashboard/framework/redis/config/RedisConfig.java +++ b/src/main/java/cn/iocoder/dashboard/framework/redis/config/RedisConfig.java @@ -1,15 +1,21 @@ package cn.iocoder.dashboard.framework.redis.config; +import cn.hutool.core.net.NetUtil; import cn.iocoder.dashboard.framework.redis.core.pubsub.AbstractChannelMessageListener; +import cn.iocoder.dashboard.framework.redis.core.stream.AbstractStreamMessageListener; import lombok.extern.slf4j.Slf4j; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.stream.*; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.listener.ChannelTopic; import org.springframework.data.redis.listener.RedisMessageListenerContainer; import org.springframework.data.redis.serializer.RedisSerializer; +import org.springframework.data.redis.stream.StreamMessageListenerContainer; +import org.springframework.util.ErrorHandler; +import java.time.Duration; import java.util.List; /** @@ -48,4 +54,52 @@ public class RedisConfig { return container; } + @Bean(initMethod = "start", destroyMethod = "stop") + public StreamMessageListenerContainer> redisStreamMessageListenerContainer( + RedisConnectionFactory factory, List> listeners) { + // 创建配置对象 + StreamMessageListenerContainer.StreamMessageListenerContainerOptions> + streamMessageListenerContainerOptions = StreamMessageListenerContainer.StreamMessageListenerContainerOptions + .builder() + // 一次性最多拉取多少条消息 + .batchSize(10) + // 执行消息轮询的执行器 + // .executor(this.threadPoolTaskExecutor) + // 消息消费异常的handler + .errorHandler(new ErrorHandler() { + @Override + public void handleError(Throwable t) { + // throw new RuntimeException(t); + t.printStackTrace(); + } + }) + // 超时时间，设置为0，表示不超时（超时后会抛出异常） + .pollTimeout(Duration.ZERO) + // 序列化器 + .serializer(RedisSerializer.string()) + .targetType(String.class) + .build(); + + // 根据配置对象创建监听容器对象 + StreamMessageListenerContainer> container = StreamMessageListenerContainer + .create(factory, streamMessageListenerContainerOptions); + + RedisTemplate redisTemplate = redisTemplate(factory); + + // 使用监听容器对象开始监听消费（使用的是手动确认方式） + String consumerName = NetUtil.getLocalHostName(); // TODO 需要优化下，晚点参考下 rocketmq consumer 的 + for (AbstractStreamMessageListener listener : listeners) { + try { + redisTemplate.opsForStream().createGroup(listener.getStreamKey(), listener.getGroup()); + } catch (Exception ignore) { +// ignore.printStackTrace(); + } + + container.receive(Consumer.from(listener.getGroup(), consumerName), + StreamOffset.create(listener.getStreamKey(), ReadOffset.lastConsumed()), listener); + } + + return container; + } + } diff --git a/src/main/java/cn/iocoder/dashboard/framework/redis/core/pubsub/ChannelMessage.java b/src/main/java/cn/iocoder/dashboard/framework/redis/core/pubsub/ChannelMessage.java index 645ae1336..93ea99f1c 100644 --- a/src/main/java/cn/iocoder/dashboard/framework/redis/core/pubsub/ChannelMessage.java +++ b/src/main/java/cn/iocoder/dashboard/framework/redis/core/pubsub/ChannelMessage.java @@ -4,6 +4,8 @@ import com.fasterxml.jackson.annotation.JsonIgnore; /** * Redis Channel Message 接口 + * + * @author 芋道源码 */ public interface ChannelMessage { @@ -12,7 +14,7 @@ public interface ChannelMessage { * * @return Channel */ - @JsonIgnore // 必须序列化 + @JsonIgnore // 避免序列化 String getChannel(); } diff --git a/src/main/java/cn/iocoder/dashboard/framework/redis/core/stream/AbstractStreamMessageListener.java b/src/main/java/cn/iocoder/dashboard/framework/redis/core/stream/AbstractStreamMessageListener.java new file mode 100644 index 000000000..6a029507c --- /dev/null +++ b/src/main/java/cn/iocoder/dashboard/framework/redis/core/stream/AbstractStreamMessageListener.java @@ -0,0 +1,83 @@ +package cn.iocoder.dashboard.framework.redis.core.stream; + +import cn.hutool.core.util.ArrayUtil; +import lombok.Getter; +import lombok.SneakyThrows; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.data.redis.connection.stream.ObjectRecord; +import org.springframework.data.redis.stream.StreamListener; +import sun.reflect.generics.reflectiveObjects.ParameterizedTypeImpl; + +import java.lang.reflect.Type; + +/** + * Redis Stream 监听器抽象类，用于实现集群消费 + * + * @param 消息类型。一定要填写噢，不然会报错 + * + * @author 芋道源码 + */ +public abstract class AbstractStreamMessageListener + implements StreamListener> { + + /** + * 消息类型 + */ + private final Class messageType; + /** + * Redis Channel + */ + @Getter + private final String streamKey; + + /** + * Redis 消费者分组，默认使用 spring.application.name 名字 + */ + @Value("spring.application.name") + @Getter + private String group; + + @SneakyThrows + protected AbstractStreamMessageListener() { + this.messageType = getMessageClass(); + this.streamKey = messageType.newInstance().getStreamKey(); + } + + @Override + public void onMessage(ObjectRecord message) { + System.out.println(message); + } + + /** + * 处理消息 + * + * @param message 消息 + */ + public abstract void onMessage(T message); + + /** + * 通过解析类上的泛型，获得消息类型 + * + * @return 消息类型 + */ + @SuppressWarnings("unchecked") + // TODO 芋艿：稍后重构 + private Class getMessageClass() { + Class targetClass = getClass(); + while (targetClass.getSuperclass() != null) { + // 如果不是 AbstractMessageListener 父类，继续向上查找 + if (targetClass.getSuperclass() != AbstractStreamMessageListener.class) { + targetClass = targetClass.getSuperclass(); + continue; + } + // 如果是 AbstractMessageListener 父类，则解析泛型 + Type[] types = ((ParameterizedTypeImpl) targetClass.getGenericSuperclass()).getActualTypeArguments(); + if (ArrayUtil.isEmpty(types)) { + throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName())); + } + return (Class) types[0]; + } + throw new IllegalStateException(String.format("类型(%s) 找不到 AbstractStreamMessageListener 父类", getClass().getName())); + } + +} diff --git a/src/main/java/cn/iocoder/dashboard/framework/redis/core/stream/StreamMessage.java b/src/main/java/cn/iocoder/dashboard/framework/redis/core/stream/StreamMessage.java new file mode 100644 index 000000000..7b0204d4a --- /dev/null +++ b/src/main/java/cn/iocoder/dashboard/framework/redis/core/stream/StreamMessage.java @@ -0,0 +1,20 @@ +package cn.iocoder.dashboard.framework.redis.core.stream; + +import com.fasterxml.jackson.annotation.JsonIgnore; + +/** + * Redis Stream Message 接口 + * + * @author 芋道源码 + */ +public interface StreamMessage { + + /** + * 获得 Redis Stream Key + * + * @return Channel + */ + @JsonIgnore // 避免序列化 + String getStreamKey(); + +} diff --git a/src/main/java/cn/iocoder/dashboard/framework/redis/core/util/RedisMessageUtils.java b/src/main/java/cn/iocoder/dashboard/framework/redis/core/util/RedisMessageUtils.java index b5cd3780b..8d01d4cf0 100644 --- a/src/main/java/cn/iocoder/dashboard/framework/redis/core/util/RedisMessageUtils.java +++ b/src/main/java/cn/iocoder/dashboard/framework/redis/core/util/RedisMessageUtils.java @@ -1,7 +1,10 @@ package cn.iocoder.dashboard.framework.redis.core.util; import cn.iocoder.dashboard.framework.redis.core.pubsub.ChannelMessage; +import cn.iocoder.dashboard.framework.redis.core.stream.StreamMessage; import cn.iocoder.dashboard.util.json.JsonUtils; +import org.springframework.data.redis.connection.stream.RecordId; +import org.springframework.data.redis.connection.stream.StreamRecords; import org.springframework.data.redis.core.RedisTemplate; /** @@ -17,8 +20,21 @@ public class RedisMessageUtils { * @param redisTemplate Redis 操作模板 * @param message 消息 */ - public static void sendChannelMessage(RedisTemplate redisTemplate, T message) { + public static void sendChannelMessage(RedisTemplate redisTemplate, T message) { redisTemplate.convertAndSend(message.getChannel(), JsonUtils.toJsonString(message)); } + /** + * 发送 Redis 消息，基于 Redis Stream 实现 + * + * @param redisTemplate Redis 操作模板 + * @param message 消息 + * @return 消息记录的编号对象 + */ + public static RecordId sendStreamMessage(RedisTemplate redisTemplate, T message) { + return redisTemplate.opsForStream().add(StreamRecords.newRecord() + .ofObject(JsonUtils.toJsonString(message)) // 设置内容 + .withStreamKey(message.getStreamKey())); // 设置 stream key + } + } diff --git a/src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/mail/SysMailSendConsumer.java b/src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/mail/SysMailSendConsumer.java new file mode 100644 index 000000000..91b796f39 --- /dev/null +++ b/src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/mail/SysMailSendConsumer.java @@ -0,0 +1,15 @@ +package cn.iocoder.dashboard.modules.system.mq.consumer.mail; + +import cn.iocoder.dashboard.framework.redis.core.stream.AbstractStreamMessageListener; +import cn.iocoder.dashboard.modules.system.mq.message.mail.SysMailSendMessage; +import org.springframework.stereotype.Component; + +@Component +public class SysMailSendConsumer extends AbstractStreamMessageListener { + + @Override + public void onMessage(SysMailSendMessage message) { + + } + +} diff --git a/src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/sms/SysSmsSendConsumer.java b/src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/sms/SysSmsSendConsumer.java new file mode 100644 index 000000000..3dde73618 --- /dev/null +++ b/src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/sms/SysSmsSendConsumer.java @@ -0,0 +1,15 @@ +package cn.iocoder.dashboard.modules.system.mq.consumer.sms; + +import cn.iocoder.dashboard.framework.redis.core.stream.AbstractStreamMessageListener; +import cn.iocoder.dashboard.modules.system.mq.message.sms.SysSmsSendMessage; +import org.springframework.stereotype.Component; + +@Component +public class SysSmsSendConsumer extends AbstractStreamMessageListener { + + @Override + public void onMessage(SysSmsSendMessage message) { + System.out.println(message); + } + +} diff --git a/src/main/java/cn/iocoder/dashboard/modules/system/mq/message/mail/SysMailSendMessage.java b/src/main/java/cn/iocoder/dashboard/modules/system/mq/message/mail/SysMailSendMessage.java new file mode 100644 index 000000000..c9f5d2aae --- /dev/null +++ b/src/main/java/cn/iocoder/dashboard/modules/system/mq/message/mail/SysMailSendMessage.java @@ -0,0 +1,46 @@ +package cn.iocoder.dashboard.modules.system.mq.message.mail; + +import cn.iocoder.dashboard.framework.redis.core.stream.StreamMessage; +import lombok.Data; + +import javax.validation.constraints.NotNull; +import java.util.Map; + +/** + * 邮箱发送消息 + * + * @author 芋道源码 + */ +@Data +public class SysMailSendMessage implements StreamMessage { + + /** + * 邮箱地址 + */ + @NotNull(message = "邮箱地址不能为空") + private String address; + /** + * 短信模板编号 + */ + @NotNull(message = "短信模板编号不能为空") + private String templateCode; + /** + * 短信模板参数 + */ + private Map templateParams; + + /** + * 用户编号，允许空 + */ + private Integer userId; + /** + * 用户类型，允许空 + */ + private Integer userType; + + @Override + public String getStreamKey() { + return "system.mail.send"; + } + +} diff --git a/src/main/java/cn/iocoder/dashboard/modules/system/mq/message/sms/SysSmsSendMessage.java b/src/main/java/cn/iocoder/dashboard/modules/system/mq/message/sms/SysSmsSendMessage.java new file mode 100644 index 000000000..f47b52466 --- /dev/null +++ b/src/main/java/cn/iocoder/dashboard/modules/system/mq/message/sms/SysSmsSendMessage.java @@ -0,0 +1,46 @@ +package cn.iocoder.dashboard.modules.system.mq.message.sms; + +import cn.iocoder.dashboard.framework.redis.core.stream.StreamMessage; +import lombok.Data; + +import javax.validation.constraints.NotNull; +import java.util.Map; + +/** + * 短信发送消息 + * + * @author 芋道源码 + */ +@Data +public class SysSmsSendMessage implements StreamMessage { + + /** + * 手机号 + */ + @NotNull(message = "手机号不能为空") + private String mobile; + /** + * 短信模板编号 + */ + @NotNull(message = "短信模板编号不能为空") + private String templateCode; + /** + * 短信模板参数 + */ + private Map templateParams; + + /** + * 用户编号，允许空 + */ + private Integer userId; + /** + * 用户类型，允许空 + */ + private Integer userType; + + @Override + public String getStreamKey() { + return "system.sms.send"; + } + +} diff --git a/src/test-integration/java/cn/iocoder/dashboard/BaseRedisIntegrationTest.java b/src/test-integration/java/cn/iocoder/dashboard/BaseRedisIntegrationTest.java new file mode 100644 index 000000000..e32eb249e --- /dev/null +++ b/src/test-integration/java/cn/iocoder/dashboard/BaseRedisIntegrationTest.java @@ -0,0 +1,23 @@ +package cn.iocoder.dashboard; + +import cn.iocoder.dashboard.framework.redis.config.RedisConfig; +import org.redisson.spring.starter.RedissonAutoConfiguration; +import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.context.annotation.Import; +import org.springframework.test.context.ActiveProfiles; + +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, classes = BaseRedisIntegrationTest.Application.class) +@ActiveProfiles("integration-test") // 设置使用 application-integration-test 配置文件 +public class BaseRedisIntegrationTest { + + @Import({ + // Redis 配置类 + RedisAutoConfiguration.class, // Spring Redis 自动配置类 + RedisConfig.class, // 自己的 Redis 配置类 + RedissonAutoConfiguration.class, // Redisson 自动高配置类 + }) + public static class Application { + } + +} diff --git a/src/test-integration/java/cn/iocoder/dashboard/framework/redis/core/stream/RedisStreamTest.java b/src/test-integration/java/cn/iocoder/dashboard/framework/redis/core/stream/RedisStreamTest.java new file mode 100644 index 000000000..022f45bfd --- /dev/null +++ b/src/test-integration/java/cn/iocoder/dashboard/framework/redis/core/stream/RedisStreamTest.java @@ -0,0 +1,54 @@ +package cn.iocoder.dashboard.framework.redis.core.stream; + +import cn.hutool.core.thread.ThreadUtil; +import cn.iocoder.dashboard.BaseRedisIntegrationTest; +import cn.iocoder.dashboard.framework.redis.core.util.RedisMessageUtils; +import cn.iocoder.dashboard.modules.system.mq.consumer.mail.SysMailSendConsumer; +import cn.iocoder.dashboard.modules.system.mq.consumer.sms.SysSmsSendConsumer; +import cn.iocoder.dashboard.modules.system.mq.message.mail.SysMailSendMessage; +import cn.iocoder.dashboard.modules.system.mq.message.sms.SysSmsSendMessage; +import org.junit.jupiter.api.Test; +import org.springframework.context.annotation.Import; +import org.springframework.data.redis.core.StringRedisTemplate; + +import javax.annotation.Resource; +import java.util.concurrent.TimeUnit; + +public class RedisStreamTest { + + @Import({SysSmsSendConsumer.class, SysMailSendConsumer.class}) + public static class ConsumerTest extends BaseRedisIntegrationTest { + + @Test + public void testConsumer() { + ThreadUtil.sleep(1, TimeUnit.DAYS); + } + + } + + public static class ProducerTest extends BaseRedisIntegrationTest { + + @Resource + private StringRedisTemplate stringRedisTemplate; + + @Test + public void testProducer01() { + // 创建消息 + SysSmsSendMessage message = new SysSmsSendMessage(); + message.setMobile("15601691300").setTemplateCode("test"); + // 发送消息 + RedisMessageUtils.sendStreamMessage(stringRedisTemplate, message); + } + + @Test + public void testProducer02() { + // 创建消息 + SysMailSendMessage message = new SysMailSendMessage(); + message.setAddress("fangfang@mihayou.com").setTemplateCode("test"); + // 发送消息 + RedisMessageUtils.sendStreamMessage(stringRedisTemplate, message); + } + + } + +} diff --git a/src/test-integration/resources/application-integration-test.yaml b/src/test-integration/resources/application-integration-test.yaml new file mode 100644 index 000000000..88b92273c --- /dev/null +++ b/src/test-integration/resources/application-integration-test.yaml @@ -0,0 +1,82 @@ +spring: + main: + lazy-initialization: true # 开启懒加载，加快速度 + banner-mode: off # 单元测试，禁用 Banner + +--- #################### 数据库相关配置 #################### + +spring: + # 数据源配置项 + datasource: + name: ruoyi-vue-pro + url: jdbc:h2:mem:testdb;MODE=MYSQL;DATABASE_TO_UPPER=false; # MODE 使用 MySQL 模式；DATABASE_TO_UPPER 配置表和字段使用小写 + driver-class-name: org.h2.Driver + username: sa + password: + schema: classpath:sql/create_tables.sql # MySQL 转 H2 的语句，使用 https://www.jooq.org/translate/ 工具 + druid: + async-init: true # 单元测试，异步初始化 Druid 连接池，提升启动速度 + initial-size: 1 # 单元测试，配置为 1，提升启动速度 + + # Redis 配置。Redisson 默认的配置足够使用，一般不需要进行调优 + redis: + host: 127.0.0.1 # 地址 + port: 6379 # 端口（单元测试，使用 16379 端口） + database: 0 # 数据库索引 + +mybatis: + lazy-initialization: true # 单元测试，设置 MyBatis Mapper 延迟加载，加速每个单元测试 + +--- #################### 定时任务相关配置 #################### + +--- #################### 配置中心相关配置 #################### + +--- #################### 服务保障相关配置 #################### + +# Lock4j 配置项（单元测试，禁用 Lock4j） + +# Resilience4j 配置项 +resilience4j: + ratelimiter: + instances: + backendA: + limit-for-period: 1 # 每个周期内，允许的请求数。默认为 50 + limit-refresh-period: 60s # 每个周期的时长，单位：微秒。默认为 500 + timeout-duration: 1s # 被限流时，阻塞等待的时长，单位：微秒。默认为 5s + register-health-indicator: true # 是否注册到健康监测 + +--- #################### 监控相关配置 #################### + +--- #################### 芋道相关配置 #################### + +# 芋道配置项，设置当前项目所有自定义的配置 +yudao: + info: + version: 1.0.0 + base-package: cn.iocoder.dashboard + web: + api-prefix: /api + controller-package: ${yudao.info.base-package} + security: + token-header: Authorization + token-secret: abcdefghijklmnopqrstuvwxyz + token-timeout: 1d + session-timeout: 30m + mock-enable: true + mock-secret: test + swagger: + enable: false # 单元测试，禁用 Swagger + captcha: + timeout: 5m + width: 160 + height: 60 + file: + base-path: http://127.0.0.1:${server.port}/${yudao.web.api-prefix}/file/get/ + codegen: + base-package: ${yudao.info.base-package}.modules + db-schemas: ${spring.datasource.name} + xss: + enable: false + exclude-urls: # 如下两个 url，仅仅是为了演示，去掉配置也没关系 + - ${spring.boot.admin.context-path}/** # 不处理 Spring Boot Admin 的请求 + - ${management.endpoints.web.base-path}/** # 不处理 Actuator 的请求 diff --git a/src/test/java/cn/iocoder/dashboard/modules/system/service/dept/SysDeptServiceTest.java b/src/test/java/cn/iocoder/dashboard/modules/system/service/dept/SysDeptServiceTest.java index 727d08f78..373665546 100644 --- a/src/test/java/cn/iocoder/dashboard/modules/system/service/dept/SysDeptServiceTest.java +++ b/src/test/java/cn/iocoder/dashboard/modules/system/service/dept/SysDeptServiceTest.java @@ -72,7 +72,6 @@ class SysDeptServiceTest extends BaseDbUnitTest { // 断言 maxUpdateTime 缓存 Date maxUpdateTime = (Date) getFieldValue(deptService, "maxUpdateTime"); assertEquals(ObjectUtils.max(deptDO1.getUpdateTime(), deptDO2.getUpdateTime()), maxUpdateTime); - } @Test