Merge branch 'master' of gitee.com:zhijiantianya/ruoyi-vue-pro into feature/ut-auth

neilz
2 parents 52d4d9fd d6cc9e23
Showing 17 changed files with 498 additions and 51 deletions
pom.xml
src/main/java/cn/iocoder/dashboard/framework/redis/config/RedisConfig.java
src/main/java/cn/iocoder/dashboard/framework/redis/core/pubsub/AbstractChannelMessageListener.java
src/main/java/cn/iocoder/dashboard/framework/redis/core/pubsub/ChannelMessage.java
src/main/java/cn/iocoder/dashboard/framework/redis/core/stream/AbstractStreamMessageListener.java
src/main/java/cn/iocoder/dashboard/framework/redis/core/stream/StreamMessage.java
src/main/java/cn/iocoder/dashboard/framework/redis/core/util/RedisMessageUtils.java
src/main/java/cn/iocoder/dashboard/modules/system/dal/mysql/permission/SysRoleMenuMapper.java
src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/mail/SysMailSendConsumer.java
src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/sms/SysSmsSendConsumer.java
src/main/java/cn/iocoder/dashboard/modules/system/mq/message/mail/SysMailSendMessage.java
src/main/java/cn/iocoder/dashboard/modules/system/mq/message/sms/SysSmsSendMessage.java
src/main/java/cn/iocoder/dashboard/modules/system/service/permission/impl/SysPermissionServiceImpl.java
src/test-integration/java/cn/iocoder/dashboard/BaseRedisIntegrationTest.java
src/test-integration/java/cn/iocoder/dashboard/framework/redis/core/stream/RedisStreamTest.java
src/test-integration/resources/application-integration-test.yaml
src/test/java/cn/iocoder/dashboard/modules/system/service/dept/SysDeptServiceTest.java
@@ -22,15 +22,15 @@
         <maven.compiler.target>${java.version}</maven.compiler.target>
         <maven-compiler-plugin.version>3.8.0</maven-compiler-plugin.version>
         <!-- 统一依赖管理 -->
-        <spring.boot.version>2.4.2</spring.boot.version>
+        <spring.boot.version>2.4.4</spring.boot.version>
         <!-- Web 相关 -->
         <knife4j.version>3.0.2</knife4j.version>
         <swagger-annotations.version>1.5.22</swagger-annotations.version>
         <!-- DB 相关 -->
         <mysql-connector-java.version>5.1.46</mysql-connector-java.version>
         <druid.version>1.2.4</druid.version>
-        <mybatis-plus.version>3.4.1</mybatis-plus.version>
-        <redisson.version>3.14.1</redisson.version>
+        <mybatis-plus.version>3.4.2</mybatis-plus.version>
+        <redisson.version>3.15.1</redisson.version>
         <!-- Config 配置中心相关 -->
         <apollo.version>1.7.0</apollo.version>
         <!-- 服务保障相关 -->
@@ -42,7 +42,7 @@
         <!-- 工具类相关 -->
         <lombok.version>1.16.14</lombok.version>
         <mapstruct.version>1.4.1.Final</mapstruct.version>
-        <hutool.version>5.5.6</hutool.version>
+        <hutool.version>5.6.1</hutool.version>
         <easyexcel.verion>2.2.7</easyexcel.verion>
         <velocity.version>2.2</velocity.version>
         <screw.version>1.0.5</screw.version>
@@ -249,27 +249,7 @@
         <dependency>
             <groupId>cn.hutool</groupId>
-            <artifactId>hutool-core</artifactId>
-            <version>${hutool.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>cn.hutool</groupId>
-            <artifactId>hutool-extra</artifactId>
-            <version>${hutool.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>cn.hutool</groupId>
-            <artifactId>hutool-captcha</artifactId>
-            <version>${hutool.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>cn.hutool</groupId>
-            <artifactId>hutool-http</artifactId>
-            <version>${hutool.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>cn.hutool</groupId>
-            <artifactId>hutool-crypto</artifactId>
+            <artifactId>hutool-all</artifactId>
             <version>${hutool.version}</version>
         </dependency>
 package cn.iocoder.dashboard.framework.redis.config;
+import cn.hutool.system.SystemUtil;
 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.Consumer;
+import org.springframework.data.redis.connection.stream.ObjectRecord;
+import org.springframework.data.redis.connection.stream.ReadOffset;
+import org.springframework.data.redis.connection.stream.StreamOffset;
 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 java.util.List;
@@ -19,6 +26,9 @@ import java.util.List;
 @Slf4j
 public class RedisConfig {
+    /**
+     * 创建 RedisTemplate Bean，使用 JSON 序列化方式
+     */
     @Bean
     public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
         // 创建 RedisTemplate 对象
@@ -27,11 +37,16 @@ public class RedisConfig {
         template.setConnectionFactory(factory);
         // 使用 String 序列化方式，序列化 KEY 。
         template.setKeySerializer(RedisSerializer.string());
+        template.setHashKeySerializer(RedisSerializer.string());
         // 使用 JSON 序列化方式（库是 Jackson ），序列化 VALUE 。
         template.setValueSerializer(RedisSerializer.json());
+        template.setHashValueSerializer(RedisSerializer.json());
         return template;
     }
+    /**
+     * 创建 Redis Pub/Sub 广播消费的容器
+     */
     @Bean
     public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory factory,
                                                                        List<AbstractChannelMessageListener<?>> listeners) {
@@ -48,4 +63,51 @@ public class RedisConfig {
         return container;
     }
+    /**
+     * 创建 Redis Stream 集群消费的容器
+     *
+     * Redis Stream 的 xreadgroup 命令：https://www.geek-book.com/src/docs/redis/redis/redis.io/commands/xreadgroup.html
+     */
+    @Bean(initMethod = "start", destroyMethod = "stop")
+    public StreamMessageListenerContainer<String, ObjectRecord<String, String>> redisStreamMessageListenerContainer(RedisTemplate<String, Object> redisTemplate,
+                                                                                                                    List<AbstractStreamMessageListener<?>> listeners) {
+        // 第一步，创建 StreamMessageListenerContainer 容器
+        // 创建 options 配置
+        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>> containerOptions =
+                StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
+                        .batchSize(10) // 一次性最多拉取多少条消息
+                        .targetType(String.class) // 目标类型。统一使用 String，通过自己封装的 AbstractStreamMessageListener 去反序列化
+                        .build();
+        // 创建 container 对象
+        StreamMessageListenerContainer<String, ObjectRecord<String, String>> container = StreamMessageListenerContainer.create(
+                redisTemplate.getRequiredConnectionFactory(), containerOptions);
+
+        // 第二步，注册监听器，消费对应的 Stream 主题
+//        String consumerName = buildConsumerName();
+        String consumerName = "110";
+        listeners.forEach(listener -> {
+            // 创建 listener 对应的消费者分组
+            try {
+                redisTemplate.opsForStream().createGroup(listener.getStreamKey(), listener.getGroup());
+            } catch (Exception ignore) {}
+            // 设置 listener 对应的 redisTemplate
+            listener.setRedisTemplate(redisTemplate);
+            // 创建 Consumer 对象
+            Consumer consumer = Consumer.from(listener.getGroup(), consumerName);
+            // 设置 Consumer 消费进度，以最小消费进度为准
+            StreamOffset<String> streamOffset = StreamOffset.create(listener.getStreamKey(), ReadOffset.lastConsumed());
+            // 设置 Consumer 监听
+            StreamMessageListenerContainer.StreamReadRequestBuilder<String> builder = StreamMessageListenerContainer.StreamReadRequest
+                    .builder(streamOffset).consumer(consumer)
+                    .autoAcknowledge(false) // 不自动 ack
+                    .cancelOnError(throwable -> false); // 默认配置，发生异常就取消消费，显然不符合预期；因此，我们设置为 false
+            container.register(builder.build(), listener);
+        });
+        return container;
+    }
+
+    private static String buildConsumerName() {
+        return String.format("%s@%d", SystemUtil.getHostInfo().getAddress(), SystemUtil.getCurrentPID());
+    }
+
 }
 package cn.iocoder.dashboard.framework.redis.core.pubsub;
-import cn.hutool.core.util.ArrayUtil;
+import cn.hutool.core.util.TypeUtil;
 import cn.iocoder.dashboard.util.json.JsonUtils;
 import lombok.SneakyThrows;
 import org.springframework.data.redis.connection.Message;
 import org.springframework.data.redis.connection.MessageListener;
-import sun.reflect.generics.reflectiveObjects.ParameterizedTypeImpl;
 import java.lang.reflect.Type;
@@ -62,21 +61,11 @@ public abstract class AbstractChannelMessageListener&lt;T extends ChannelMessage&gt; i
      */
     @SuppressWarnings("unchecked")
     private Class<T> getMessageClass() {
-        Class<?> targetClass = getClass();
-        while (targetClass.getSuperclass() != null) {
-            // 如果不是 AbstractMessageListener 父类，继续向上查找
-            if (targetClass.getSuperclass() != AbstractChannelMessageListener.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<T>) types[0];
+        Type type = TypeUtil.getTypeArgument(getClass(), 0);
+        if (type == null) {
+            throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName()));
         }
-        throw new IllegalStateException(String.format("类型(%s) 找不到 AbstractMessageListener 父类", getClass().getName()));
+        return (Class<T>) type;
     }
 }
@@ -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();
 }
+package cn.iocoder.dashboard.framework.redis.core.stream;
+
+import cn.hutool.core.util.TypeUtil;
+import cn.iocoder.dashboard.util.json.JsonUtils;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.SneakyThrows;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.redis.connection.stream.ObjectRecord;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.stream.StreamListener;
+
+import java.lang.reflect.Type;
+
+/**
+ * Redis Stream 监听器抽象类，用于实现集群消费
+ *
+ * @param <T> 消息类型。一定要填写噢，不然会报错
+ *
+ * @author 芋道源码
+ */
+public abstract class AbstractStreamMessageListener<T extends StreamMessage>
+        implements StreamListener<String, ObjectRecord<String, String>> {
+
+    /**
+     * 消息类型
+     */
+    private final Class<T> messageType;
+    /**
+     * Redis Channel
+     */
+    @Getter
+    private final String streamKey;
+
+    /**
+     * Redis 消费者分组，默认使用 spring.application.name 名字
+     */
+    @Value("${spring.application.name}")
+    @Getter
+    private String group;
+    /**
+     *
+     */
+    @Setter
+    private RedisTemplate<String, ?> redisTemplate;
+
+    @SneakyThrows
+    protected AbstractStreamMessageListener() {
+        this.messageType = getMessageClass();
+        this.streamKey = messageType.newInstance().getStreamKey();
+    }
+
+    @Override
+    public void onMessage(ObjectRecord<String, String> message) {
+        // 消费消息
+        T messageObj = JsonUtils.parseObject(message.getValue(), messageType);
+        this.onMessage(messageObj);
+        // ack 消息消费完成
+        redisTemplate.opsForStream().acknowledge(group, message);
+        // TODO 芋艿：需要额外考虑以下几个点：
+        // 1. 处理异常的情况
+        // 2. 发送日志；以及事务的结合
+        // 3. 消费日志；以及通用的幂等性
+        // 4. 消费失败的重试，https://zhuanlan.zhihu.com/p/60501638
+    }
+
+    /**
+     * 处理消息
+     *
+     * @param message 消息
+     */
+    public abstract void onMessage(T message);
+
+    /**
+     * 通过解析类上的泛型，获得消息类型
+     *
+     * @return 消息类型
+     */
+    @SuppressWarnings("unchecked")
+    private Class<T> getMessageClass() {
+        Type type = TypeUtil.getTypeArgument(getClass(), 0);
+        if (type == null) {
+            throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName()));
+        }
+        return (Class<T>) type;
+    }
+
+}
+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();
+
+}
 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 <T extends ChannelMessage>  void sendChannelMessage(RedisTemplate<?, ?> redisTemplate, T message) {
+    public static <T extends ChannelMessage> void sendChannelMessage(RedisTemplate<?, ?> redisTemplate, T message) {
         redisTemplate.convertAndSend(message.getChannel(), JsonUtils.toJsonString(message));
     }
+    /**
+     * 发送 Redis 消息，基于 Redis Stream 实现
+     *
+     * @param redisTemplate Redis 操作模板
+     * @param message 消息
+     * @return 消息记录的编号对象
+     */
+    public static <T extends StreamMessage> RecordId sendStreamMessage(RedisTemplate<String, ?> redisTemplate, T message) {
+        return redisTemplate.opsForStream().add(StreamRecords.newRecord()
+                .ofObject(JsonUtils.toJsonString(message)) // 设置内容
+                .withStreamKey(message.getStreamKey())); // 设置 stream key
+    }
+
 }
@@ -4,6 +4,7 @@ import cn.iocoder.dashboard.framework.mybatis.core.mapper.BaseMapperX;
 import cn.iocoder.dashboard.modules.system.dal.dataobject.permission.SysRoleMenuDO;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Select;
 import java.util.Collection;
 import java.util.Date;
@@ -32,7 +33,7 @@ public interface SysRoleMenuMapper extends BaseMapperX&lt;SysRoleMenuDO&gt; {
         delete(new QueryWrapper<SysRoleMenuDO>().eq("role_id", roleId)
                 .in("menu_id", menuIds));
     }
-
+    
     default void deleteListByMenuId(Long menuId) {
         delete(new QueryWrapper<SysRoleMenuDO>().eq("menu_id", menuId));
     }
@@ -41,9 +42,7 @@ public interface SysRoleMenuMapper extends BaseMapperX&lt;SysRoleMenuDO&gt; {
         delete(new QueryWrapper<SysRoleMenuDO>().eq("role_id", roleId));
     }
-    default boolean selectExistsByUpdateTimeAfter(Date maxUpdateTime) {
-        return selectOne(new QueryWrapper<SysRoleMenuDO>().select("id")
-                .gt("update_time", maxUpdateTime).last("LIMIT 1")) != null;
-    }
+    @Select("SELECT id FROM sys_role_menu WHERE update_time > #{maxUpdateTime} LIMIT 1")
+    Long selectExistsByUpdateTimeAfter(Date maxUpdateTime);
 }
+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 lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Component
+@Slf4j
+public class SysMailSendConsumer extends AbstractStreamMessageListener<SysMailSendMessage> {
+
+    @Override
+    public void onMessage(SysMailSendMessage message) {
+        log.info("[onMessage][消息内容({})]", message);
+    }
+
+}
+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 lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Component
+@Slf4j
+public class SysSmsSendConsumer extends AbstractStreamMessageListener<SysSmsSendMessage> {
+
+    @Override
+    public void onMessage(SysSmsSendMessage message) {
+        log.info("[onMessage][消息内容({})]", message);
+    }
+
+}
+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<String, Object> templateParams;
+
+    /**
+     * 用户编号，允许空
+     */
+    private Integer userId;
+    /**
+     * 用户类型，允许空
+     */
+    private Integer userType;
+
+    @Override
+    public String getStreamKey() {
+        return "system.mail.send";
+    }
+
+}
+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<String, Object> templateParams;
+
+    /**
+     * 用户编号，允许空
+     */
+    private Integer userId;
+    /**
+     * 用户类型，允许空
+     */
+    private Integer userType;
+
+    @Override
+    public String getStreamKey() {
+        return "system.sms.send";
+    }
+
+}
@@ -86,6 +86,7 @@ public class SysPermissionServiceImpl implements SysPermissionService {
     @Override
     @PostConstruct
     public void initLocalCache() {
+        Date now = new Date();
         // 获取角色与菜单的关联列表，如果有更新
         List<SysRoleMenuDO> roleMenuList = this.loadRoleMenuIfUpdate(maxUpdateTime);
         if (CollUtil.isEmpty(roleMenuList)) {
@@ -102,7 +103,7 @@ public class SysPermissionServiceImpl implements SysPermissionService {
         roleMenuCache = roleMenuCacheBuilder.build();
         menuRoleCache = menuRoleCacheBuilder.build();
         assert roleMenuList.size() > 0; // 断言，避免告警
-        maxUpdateTime = roleMenuList.stream().max(Comparator.comparing(BaseDO::getUpdateTime)).get().getUpdateTime();
+        maxUpdateTime = now;
         log.info("[initLocalCache][初始化角色与菜单的关联数量为 {}]", roleMenuList.size());
     }
@@ -123,7 +124,7 @@ public class SysPermissionServiceImpl implements SysPermissionService {
         if (maxUpdateTime == null) { // 如果更新时间为空，说明 DB 一定有新数据
             log.info("[loadRoleMenuIfUpdate][首次加载全量角色与菜单的关联]");
         } else { // 判断数据库中是否有更新的角色与菜单的关联
-            if (!roleMenuMapper.selectExistsByUpdateTimeAfter(maxUpdateTime)) {
+            if (Objects.isNull(roleMenuMapper.selectExistsByUpdateTimeAfter(maxUpdateTime))) {
                 return null;
             }
             log.info("[loadRoleMenuIfUpdate][增量加载全量角色与菜单的关联]");
+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 {
+    }
+
+}
+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.RedisTemplate;
+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;
+
+        @Resource
+        private RedisTemplate<String, Object> redisTemplate;
+
+        @Test
+        public void testProducer01() {
+            for (int i = 0; i < 100; i++) {
+                // 创建消息
+                SysSmsSendMessage message = new SysSmsSendMessage();
+                message.setMobile("15601691300").setTemplateCode("test:" + i);
+                // 发送消息
+                RedisMessageUtils.sendStreamMessage(stringRedisTemplate, message);
+            }
+        }
+
+        @Test
+        public void testProducer02() {
+            // 创建消息
+            SysMailSendMessage message = new SysMailSendMessage();
+            message.setAddress("fangfang@mihayou.com").setTemplateCode("test");
+            // 发送消息
+            RedisMessageUtils.sendStreamMessage(stringRedisTemplate, message);
+        }
+
+    }
+
+}
+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 的请求
@@ -72,7 +72,6 @@ class SysDeptServiceTest extends BaseDbUnitTest {
         // 断言 maxUpdateTime 缓存
         Date maxUpdateTime = (Date) getFieldValue(deptService, "maxUpdateTime");
         assertEquals(ObjectUtils.max(deptDO1.getUpdateTime(), deptDO2.getUpdateTime()), maxUpdateTime);
-
     }
     @Test