diff --git a/pom.xml b/pom.xml
index bba8d18cc..96e0e08ce 100644
--- a/pom.xml
+++ b/pom.xml
@@ -22,15 +22,15 @@
${java.version}
3.8.0
- 2.4.2
+ 2.4.4
3.0.2
1.5.22
5.1.46
1.2.4
- 3.4.1
- 3.14.1
+ 3.4.2
+ 3.15.1
1.7.0
@@ -42,7 +42,7 @@
1.16.14
1.4.1.Final
- 5.5.6
+ 5.6.1
2.2.7
2.2
1.0.5
@@ -249,27 +249,7 @@
cn.hutool
- hutool-core
- ${hutool.version}
-
-
- cn.hutool
- hutool-extra
- ${hutool.version}
-
-
- cn.hutool
- hutool-captcha
- ${hutool.version}
-
-
- cn.hutool
- hutool-http
- ${hutool.version}
-
-
- cn.hutool
- hutool-crypto
+ hutool-all
${hutool.version}
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..f398625ea 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,14 +1,21 @@
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 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> 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> redisStreamMessageListenerContainer(RedisTemplate redisTemplate,
+ List> listeners) {
+ // 第一步，创建 StreamMessageListenerContainer 容器
+ // 创建 options 配置
+ StreamMessageListenerContainer.StreamMessageListenerContainerOptions> containerOptions =
+ StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
+ .batchSize(10) // 一次性最多拉取多少条消息
+ .targetType(String.class) // 目标类型。统一使用 String，通过自己封装的 AbstractStreamMessageListener 去反序列化
+ .build();
+ // 创建 container 对象
+ StreamMessageListenerContainer> 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 streamOffset = StreamOffset.create(listener.getStreamKey(), ReadOffset.lastConsumed());
+ // 设置 Consumer 监听
+ StreamMessageListenerContainer.StreamReadRequestBuilder 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());
+ }
+
}
diff --git a/src/main/java/cn/iocoder/dashboard/framework/redis/core/pubsub/AbstractChannelMessageListener.java b/src/main/java/cn/iocoder/dashboard/framework/redis/core/pubsub/AbstractChannelMessageListener.java
index 23b40e228..2abf03d4d 100644
--- a/src/main/java/cn/iocoder/dashboard/framework/redis/core/pubsub/AbstractChannelMessageListener.java
+++ b/src/main/java/cn/iocoder/dashboard/framework/redis/core/pubsub/AbstractChannelMessageListener.java
@@ -1,11 +1,10 @@
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 i
*/
@SuppressWarnings("unchecked")
private Class 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) 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) type;
}
}
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..5fb01aa15
--- /dev/null
+++ b/src/main/java/cn/iocoder/dashboard/framework/redis/core/stream/AbstractStreamMessageListener.java
@@ -0,0 +1,88 @@
+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 消息类型。一定要填写噢，不然会报错
+ *
+ * @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;
+ /**
+ *
+ */
+ @Setter
+ private RedisTemplate redisTemplate;
+
+ @SneakyThrows
+ protected AbstractStreamMessageListener() {
+ this.messageType = getMessageClass();
+ this.streamKey = messageType.newInstance().getStreamKey();
+ }
+
+ @Override
+ public void onMessage(ObjectRecord 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 getMessageClass() {
+ Type type = TypeUtil.getTypeArgument(getClass(), 0);
+ if (type == null) {
+ throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName()));
+ }
+ return (Class) type;
+ }
+
+}
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..9331606af 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/dal/mysql/permission/SysRoleMenuMapper.java b/src/main/java/cn/iocoder/dashboard/modules/system/dal/mysql/permission/SysRoleMenuMapper.java
index a28d546e4..4a7dcda70 100644
--- a/src/main/java/cn/iocoder/dashboard/modules/system/dal/mysql/permission/SysRoleMenuMapper.java
+++ b/src/main/java/cn/iocoder/dashboard/modules/system/dal/mysql/permission/SysRoleMenuMapper.java
@@ -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 {
delete(new QueryWrapper().eq("role_id", roleId)
.in("menu_id", menuIds));
}
-
+
default void deleteListByMenuId(Long menuId) {
delete(new QueryWrapper().eq("menu_id", menuId));
}
@@ -41,9 +42,7 @@ public interface SysRoleMenuMapper extends BaseMapperX {
delete(new QueryWrapper().eq("role_id", roleId));
}
- default boolean selectExistsByUpdateTimeAfter(Date maxUpdateTime) {
- return selectOne(new QueryWrapper().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);
}
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..3a0f22ee6
--- /dev/null
+++ b/src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/mail/SysMailSendConsumer.java
@@ -0,0 +1,17 @@
+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 {
+
+ @Override
+ public void onMessage(SysMailSendMessage message) {
+ log.info("[onMessage][消息内容({})]", 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..e3b18ca75
--- /dev/null
+++ b/src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/sms/SysSmsSendConsumer.java
@@ -0,0 +1,17 @@
+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 {
+
+ @Override
+ public void onMessage(SysSmsSendMessage message) {
+ log.info("[onMessage][消息内容({})]", 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/main/java/cn/iocoder/dashboard/modules/system/service/permission/impl/SysPermissionServiceImpl.java b/src/main/java/cn/iocoder/dashboard/modules/system/service/permission/impl/SysPermissionServiceImpl.java
index 0e1b053da..ba2a096b6 100644
--- a/src/main/java/cn/iocoder/dashboard/modules/system/service/permission/impl/SysPermissionServiceImpl.java
+++ b/src/main/java/cn/iocoder/dashboard/modules/system/service/permission/impl/SysPermissionServiceImpl.java
@@ -86,6 +86,7 @@ public class SysPermissionServiceImpl implements SysPermissionService {
@Override
@PostConstruct
public void initLocalCache() {
+ Date now = new Date();
// 获取角色与菜单的关联列表，如果有更新
List 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][增量加载全量角色与菜单的关联]");
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..3d0d8a249
--- /dev/null
+++ b/src/test-integration/java/cn/iocoder/dashboard/framework/redis/core/stream/RedisStreamTest.java
@@ -0,0 +1,60 @@
+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 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);
+ }
+
+ }
+
+}
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