From 8683401c802e1e98d9de8f5427a8f1cdf4b0c4a5 Mon Sep 17 00:00:00 2001 From: dxyx <5676377+dxyx@user.noreply.gitee.com> Date: Fri, 12 Mar 2021 09:31:15 +0800 Subject: [PATCH 1/6] =?UTF-8?q?=E8=A7=A3=E5=86=B3=E7=94=B1=E4=BA=8Emybatis?= =?UTF-8?q?-plus=E6=97=A0=E6=B3=95=E8=BF=87=E6=BB=A4=E8=BD=AF=E5=88=A0?= =?UTF-8?q?=E9=99=A4=E5=AF=BC=E8=87=B4=E8=A7=92=E8=89=B2=E5=8E=BB=E6=8E=89?= =?UTF-8?q?=E8=8F=9C=E5=8D=95=E7=BC=93=E5=AD=98=E6=9C=AA=E5=88=B7=E6=96=B0?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E9=A2=98=E3=80=82=20=E8=A7=A3=E5=86=B3?= =?UTF-8?q?=E5=8E=BB=E6=8E=89=E8=8F=9C=E5=8D=95=E7=BC=93=E5=AD=98=E6=9C=AA?= =?UTF-8?q?=E5=88=B7=E6=96=B0=E7=9A=84=E9=97=AE=E9=A2=98=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../system/dal/mysql/permission/SysRoleMenuMapper.java | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) 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 b93bb5917..e6bdc56cd 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 @@ -33,9 +33,7 @@ public interface SysRoleMenuMapper extends BaseMapperX { .in("menu_id", menuIds)); } - 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") + List selectExistsByUpdateTimeAfter(Date maxUpdateTime); } From bbea33e72efc6612597e6beac0260d3b140164d0 Mon Sep 17 00:00:00 2001 From: dxyx <5676377+dxyx@user.noreply.gitee.com> Date: Fri, 12 Mar 2021 09:34:01 +0800 Subject: [PATCH 2/6] =?UTF-8?q?=E8=A7=A3=E5=86=B3=E7=94=B1=E4=BA=8Emybatis?= =?UTF-8?q?-plus=E6=97=A0=E6=B3=95=E8=BF=87=E6=BB=A4=E8=BD=AF=E5=88=A0?= =?UTF-8?q?=E9=99=A4=E5=AF=BC=E8=87=B4=E8=A7=92=E8=89=B2=E5=8E=BB=E6=8E=89?= =?UTF-8?q?=E8=8F=9C=E5=8D=95=E7=BC=93=E5=AD=98=E6=9C=AA=E5=88=B7=E6=96=B0?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E9=A2=98=E3=80=82=20=E8=A7=A3=E5=86=B3?= =?UTF-8?q?=E7=94=B1=E4=BA=8Emybatis-plus=E6=97=A0=E6=B3=95=E8=BF=87?= =?UTF-8?q?=E6=BB=A4=E8=BD=AF=E5=88=A0=E9=99=A4=E5=AF=BC=E8=87=B4=E8=A7=92?= =?UTF-8?q?=E8=89=B2=E5=8E=BB=E6=8E=89=E8=8F=9C=E5=8D=95=E7=BC=93=E5=AD=98?= =?UTF-8?q?=E6=9C=AA=E5=88=B7=E6=96=B0=E7=9A=84=E9=97=AE=E9=A2=98=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/permission/impl/SysPermissionServiceImpl.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 9f48af9f5..835fb8961 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 @@ -123,7 +123,7 @@ public class SysPermissionServiceImpl implements SysPermissionService { if (maxUpdateTime == null) { // 如果更新时间为空，说明 DB 一定有新数据 log.info("[loadRoleMenuIfUpdate][首次加载全量角色与菜单的关联]"); } else { // 判断数据库中是否有更新的角色与菜单的关联 - if (!roleMenuMapper.selectExistsByUpdateTimeAfter(maxUpdateTime)) { + if (roleMenuMapper.selectExistsByUpdateTimeAfter(maxUpdateTime).size() == 0) { return null; } log.info("[loadRoleMenuIfUpdate][增量加载全量角色与菜单的关联]"); From 49f25a88704b79a6713bbd4835a0359d5b629921 Mon Sep 17 00:00:00 2001 From: asas6559 <302811456@qq.com> Date: Wed, 17 Mar 2021 21:13:13 +0800 Subject: [PATCH 3/6] =?UTF-8?q?=E8=A7=A3=E5=86=B3=E7=94=B1=E4=BA=8Emybatis?= =?UTF-8?q?-plus=E6=97=A0=E6=B3=95=E8=BF=87=E6=BB=A4=E8=BD=AF=E5=88=A0?= =?UTF-8?q?=E9=99=A4=E5=AF=BC=E8=87=B4=E8=A7=92=E8=89=B2=E5=8E=BB=E6=8E=89?= =?UTF-8?q?=E8=8F=9C=E5=8D=95=E7=BC=93=E5=AD=98=E6=9C=AA=E5=88=B7=E6=96=B0?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../system/dal/mysql/permission/SysRoleMenuMapper.java | 5 +++-- .../service/permission/impl/SysPermissionServiceImpl.java | 5 +++-- 2 files changed, 6 insertions(+), 4 deletions(-) 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 e6bdc56cd..977243aa6 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; @@ -33,7 +34,7 @@ public interface SysRoleMenuMapper extends BaseMapperX { .in("menu_id", menuIds)); } - @Select("select id from sys_role_menu where update_time > #{maxUpdateTime} limit 1") - List selectExistsByUpdateTimeAfter(Date maxUpdateTime); + @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/service/permission/impl/SysPermissionServiceImpl.java b/src/main/java/cn/iocoder/dashboard/modules/system/service/permission/impl/SysPermissionServiceImpl.java index 835fb8961..fc66f5c32 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).size() == 0) { + if (Objects.isNull(roleMenuMapper.selectExistsByUpdateTimeAfter(maxUpdateTime))) { return null; } log.info("[loadRoleMenuIfUpdate][增量加载全量角色与菜单的关联]"); From be3fac75420232887eaad05c66ee028ec8baf52e Mon Sep 17 00:00:00 2001 From: YunaiV Date: Sat, 20 Mar 2021 20:39:01 +0800 Subject: [PATCH 4/6] =?UTF-8?q?1.=20=E7=AE=80=E5=8D=95=20redis=20stream=20?= =?UTF-8?q?=E7=9A=84=20StreamMessage=20=E5=92=8C=E5=AF=B9=E5=BA=94?= =?UTF-8?q?=E7=9A=84=E6=B6=88=E8=B4=B9=E8=80=85=202.=20=E8=B7=91=E9=80=9A?= =?UTF-8?q?=20Redis=20Stream=20=E7=9A=84=E6=B5=81=E7=A8=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../framework/redis/config/RedisConfig.java | 54 ++++++++++++ .../redis/core/pubsub/ChannelMessage.java | 4 +- .../stream/AbstractStreamMessageListener.java | 83 +++++++++++++++++++ .../redis/core/stream/StreamMessage.java | 20 +++++ .../redis/core/util/RedisMessageUtils.java | 18 +++- .../mq/consumer/mail/SysMailSendConsumer.java | 15 ++++ .../mq/consumer/sms/SysSmsSendConsumer.java | 15 ++++ .../mq/message/mail/SysMailSendMessage.java | 46 ++++++++++ .../mq/message/sms/SysSmsSendMessage.java | 46 ++++++++++ .../dashboard/BaseRedisIntegrationTest.java | 23 +++++ .../redis/core/stream/RedisStreamTest.java | 54 ++++++++++++ .../application-integration-test.yaml | 82 ++++++++++++++++++ .../service/dept/SysDeptServiceTest.java | 1 - 13 files changed, 458 insertions(+), 3 deletions(-) create mode 100644 src/main/java/cn/iocoder/dashboard/framework/redis/core/stream/AbstractStreamMessageListener.java create mode 100644 src/main/java/cn/iocoder/dashboard/framework/redis/core/stream/StreamMessage.java create mode 100644 src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/mail/SysMailSendConsumer.java create mode 100644 src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/sms/SysSmsSendConsumer.java create mode 100644 src/main/java/cn/iocoder/dashboard/modules/system/mq/message/mail/SysMailSendMessage.java create mode 100644 src/main/java/cn/iocoder/dashboard/modules/system/mq/message/sms/SysSmsSendMessage.java create mode 100644 src/test-integration/java/cn/iocoder/dashboard/BaseRedisIntegrationTest.java create mode 100644 src/test-integration/java/cn/iocoder/dashboard/framework/redis/core/stream/RedisStreamTest.java create mode 100644 src/test-integration/resources/application-integration-test.yaml 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 From 9c76fd4b69110a5ad802dd77ed89fc31d5364473 Mon Sep 17 00:00:00 2001 From: YunaiV Date: Sat, 20 Mar 2021 23:47:05 +0800 Subject: [PATCH 5/6] =?UTF-8?q?1.=20=E5=8D=87=E7=BA=A7=20springboot=20?= =?UTF-8?q?=E5=88=B0=E6=9C=80=E6=96=B0=E7=9A=84=E7=89=88=E6=9C=AC=EF=BC=8C?= =?UTF-8?q?=E8=A7=A3=E5=86=B3=20spring=20data=20redis=20=E5=AD=98=E5=82=A8?= =?UTF-8?q?=E7=9A=84=20bug=202.=20=E6=A2=B3=E7=90=86=20StreamMessageListen?= =?UTF-8?q?erContainer=20Bean=20=E7=9A=84=E5=88=9B=E5=BB=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pom.xml | 30 +----- .../framework/redis/config/RedisConfig.java | 95 ++++++++++--------- .../stream/AbstractStreamMessageListener.java | 3 + .../redis/core/util/RedisMessageUtils.java | 2 +- .../redis/core/stream/RedisStreamTest.java | 16 +++- 5 files changed, 70 insertions(+), 76 deletions(-) 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 13cda9dbb..f069738e5 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,21 +1,22 @@ package cn.iocoder.dashboard.framework.redis.config; -import cn.hutool.core.net.NetUtil; +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.*; +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 org.springframework.util.ErrorHandler; -import java.time.Duration; import java.util.List; /** @@ -25,6 +26,9 @@ import java.util.List; @Slf4j public class RedisConfig { + /** + * 创建 RedisTemplate Bean，使用 JSON 序列化方式 + */ @Bean public RedisTemplate redisTemplate(RedisConnectionFactory factory) { // 创建 RedisTemplate 对象 @@ -33,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) { @@ -54,52 +63,48 @@ 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( - 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(); + 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); - // 根据配置对象创建监听容器对象 - StreamMessageListenerContainer> container = StreamMessageListenerContainer - .create(factory, streamMessageListenerContainerOptions); - - RedisTemplate redisTemplate = redisTemplate(factory); - - // 使用监听容器对象开始监听消费（使用的是手动确认方式） - String consumerName = NetUtil.getLocalHostName(); // TODO 需要优化下，晚点参考下 rocketmq consumer 的 - for (AbstractStreamMessageListener listener : listeners) { + // 第二步，注册监听器，消费对应的 Stream 主题 + String consumerName = buildConsumerName(); + listeners.forEach(listener -> { + // 创建 listener 对应的消费者分组 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); - } - + } catch (Exception ignore) {} + // 创建 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/stream/AbstractStreamMessageListener.java b/src/main/java/cn/iocoder/dashboard/framework/redis/core/stream/AbstractStreamMessageListener.java index 6a029507c..7621d3638 100644 --- 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 @@ -46,6 +46,9 @@ public abstract class AbstractStreamMessageListener @Override public void onMessage(ObjectRecord message) { System.out.println(message); + if (true) { +// throw new IllegalStateException("测试下"); + } } /** 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 8d01d4cf0..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 @@ -31,7 +31,7 @@ public class RedisMessageUtils { * @param message 消息 * @return 消息记录的编号对象 */ - public static RecordId sendStreamMessage(RedisTemplate redisTemplate, T message) { + 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/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 index 022f45bfd..3d0d8a249 100644 --- 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 @@ -9,6 +9,7 @@ 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; @@ -31,13 +32,18 @@ public class RedisStreamTest { @Resource private StringRedisTemplate stringRedisTemplate; + @Resource + private RedisTemplate redisTemplate; + @Test public void testProducer01() { - // 创建消息 - SysSmsSendMessage message = new SysSmsSendMessage(); - message.setMobile("15601691300").setTemplateCode("test"); - // 发送消息 - RedisMessageUtils.sendStreamMessage(stringRedisTemplate, message); + for (int i = 0; i < 100; i++) { + // 创建消息 + SysSmsSendMessage message = new SysSmsSendMessage(); + message.setMobile("15601691300").setTemplateCode("test:" + i); + // 发送消息 + RedisMessageUtils.sendStreamMessage(stringRedisTemplate, message); + } } @Test From d6cc9e23a306f0d0ecb13810de7317aee1b4b010 Mon Sep 17 00:00:00 2001 From: YunaiV Date: Sun, 21 Mar 2021 00:50:33 +0800 Subject: [PATCH 6/6] =?UTF-8?q?=E5=AE=8C=E6=88=90=E5=9F=BA=E7=A1=80?= =?UTF-8?q?=E7=9A=84=20stream=20=E5=B0=81=E8=A3=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../framework/redis/config/RedisConfig.java | 5 +- .../AbstractChannelMessageListener.java | 21 ++------- .../stream/AbstractStreamMessageListener.java | 46 ++++++++++--------- .../mq/consumer/mail/SysMailSendConsumer.java | 4 +- .../mq/consumer/sms/SysSmsSendConsumer.java | 4 +- 5 files changed, 39 insertions(+), 41 deletions(-) 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 f069738e5..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 @@ -83,12 +83,15 @@ public class RedisConfig { redisTemplate.getRequiredConnectionFactory(), containerOptions); // 第二步，注册监听器，消费对应的 Stream 主题 - String consumerName = buildConsumerName(); +// 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 消费进度，以最小消费进度为准 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/stream/AbstractStreamMessageListener.java b/src/main/java/cn/iocoder/dashboard/framework/redis/core/stream/AbstractStreamMessageListener.java index 7621d3638..5fb01aa15 100644 --- 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 @@ -1,12 +1,14 @@ package cn.iocoder.dashboard.framework.redis.core.stream; -import cn.hutool.core.util.ArrayUtil; +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 sun.reflect.generics.reflectiveObjects.ParameterizedTypeImpl; import java.lang.reflect.Type; @@ -33,9 +35,14 @@ public abstract class AbstractStreamMessageListener /** * Redis 消费者分组，默认使用 spring.application.name 名字 */ - @Value("spring.application.name") + @Value("${spring.application.name}") @Getter private String group; + /** + * + */ + @Setter + private RedisTemplate redisTemplate; @SneakyThrows protected AbstractStreamMessageListener() { @@ -45,10 +52,16 @@ public abstract class AbstractStreamMessageListener @Override public void onMessage(ObjectRecord message) { - System.out.println(message); - if (true) { -// throw new IllegalStateException("测试下"); - } + // 消费消息 + 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 } /** @@ -64,23 +77,12 @@ public abstract class AbstractStreamMessageListener * @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]; + Type type = TypeUtil.getTypeArgument(getClass(), 0); + if (type == null) { + throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName())); } - throw new IllegalStateException(String.format("类型(%s) 找不到 AbstractStreamMessageListener 父类", getClass().getName())); + return (Class) type; } } 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 index 91b796f39..3a0f22ee6 100644 --- 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 @@ -2,14 +2,16 @@ 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 index 3dde73618..e3b18ca75 100644 --- 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 @@ -2,14 +2,16 @@ 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) { - System.out.println(message); + log.info("[onMessage][消息内容({})]", message); } }