From 0b5662b6989b5ae053f441adc80b48ad14fe02f9 Mon Sep 17 00:00:00 2001 From: YunaiV Date: Mon, 3 May 2021 01:49:56 +0800 Subject: [PATCH] 项目结构调整 x 20 : 拆分出独立的 yudao-spring-boot-starter-mq 组件，整理更加清晰 --- yudao-admin-server/pom.xml | 10 ++++++++++ yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/infra/mq/consumer/config/InfConfigRefreshConsumer.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/infra/mq/message/config/InfConfigRefreshMessage.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/infra/mq/producer/config/InfConfigProducer.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/dept/SysDeptRefreshConsumer.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/dict/SysDictDataRefreshConsumer.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/mail/SysMailSendConsumer.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/permission/SysMenuRefreshConsumer.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/permission/SysRoleMenuRefreshConsumer.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/permission/SysRoleRefreshConsumer.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/sms/SysSmsChannelRefreshConsumer.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/sms/SysSmsSendConsumer.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/sms/SysSmsTemplateRefreshConsumer.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/dept/SysDeptRefreshMessage.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/dict/SysDictDataRefreshMessage.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/mail/SysMailSendMessage.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/permission/SysMenuRefreshMessage.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/permission/SysRoleMenuRefreshMessage.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/permission/SysRoleRefreshMessage.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/sms/SysSmsChannelRefreshMessage.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/sms/SysSmsSendMessage.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/sms/SysSmsTemplateRefreshMessage.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/dept/SysDeptProducer.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/dict/SysDictDataProducer.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/permission/SysMenuProducer.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/permission/SysPermissionProducer.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/permission/SysRoleProducer.java | 2 +- yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/sms/SysSmsProducer.java | 2 +- yudao-admin-server/src/test-integration/java/cn/iocoder/yudao/adminserver/framework/redis/core/stream/RedisStreamTest.java | 2 +- yudao-dependencies/pom.xml | 7 +++++++ yudao-framework/pom.xml | 4 ++++ yudao-framework/yudao-spring-boot-starter-mq/pom.xml | 26 ++++++++++++++++++++++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/config/YudaoMQAutoConfiguration.java | 105 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/AbstractChannelMessageListener.java | 71 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/ChannelMessage.java | 20 ++++++++++++++++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/AbstractStreamMessageListener.java | 88 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/StreamMessage.java | 20 ++++++++++++++++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/util/RedisMessageUtils.java | 40 ++++++++++++++++++++++++++++++++++++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/package-info.java | 6 ++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/resources/META-INF/spring.factories | 2 ++ yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/config/YudaoRedisAutoConfiguration.java | 84 ------------------------------------------------------------------------------------ yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/pubsub/AbstractChannelMessageListener.java | 71 ----------------------------------------------------------------------- yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/pubsub/ChannelMessage.java | 20 -------------------- yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/stream/AbstractStreamMessageListener.java | 88 ---------------------------------------------------------------------------------------- yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/stream/StreamMessage.java | 20 -------------------- yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/util/RedisMessageUtils.java | 40 ---------------------------------------- 46 files changed, 427 insertions(+), 351 deletions(-) create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/pom.xml create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/config/YudaoMQAutoConfiguration.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/AbstractChannelMessageListener.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/ChannelMessage.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/AbstractStreamMessageListener.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/StreamMessage.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/util/RedisMessageUtils.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/package-info.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/resources/META-INF/spring.factories delete mode 100644 yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/pubsub/AbstractChannelMessageListener.java delete mode 100644 yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/pubsub/ChannelMessage.java delete mode 100644 yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/stream/AbstractStreamMessageListener.java delete mode 100644 yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/stream/StreamMessage.java delete mode 100644 yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/util/RedisMessageUtils.java diff --git a/yudao-admin-server/pom.xml b/yudao-admin-server/pom.xml index 15fc831..107b67e 100644 --- a/yudao-admin-server/pom.xml +++ b/yudao-admin-server/pom.xml @@ -65,6 +65,12 @@ yudao-spring-boot-starter-job + + + cn.iocoder.boot + yudao-spring-boot-starter-mq + + cn.iocoder.boot @@ -104,6 +110,10 @@ cn.smallbun.screw screw-core + + cn.iocoder.boot + yudao-spring-boot-starter-mq + diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/infra/mq/consumer/config/InfConfigRefreshConsumer.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/infra/mq/consumer/config/InfConfigRefreshConsumer.java index 4ab2ba4..35d3f79 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/infra/mq/consumer/config/InfConfigRefreshConsumer.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/infra/mq/consumer/config/InfConfigRefreshConsumer.java @@ -1,7 +1,7 @@ package cn.iocoder.yudao.adminserver.modules.infra.mq.consumer.config; import cn.iocoder.yudao.framework.apollo.internals.DBConfigRepository; -import cn.iocoder.yudao.framework.redis.core.pubsub.AbstractChannelMessageListener; +import cn.iocoder.yudao.framework.mq.core.pubsub.AbstractChannelMessageListener; import cn.iocoder.yudao.adminserver.modules.infra.mq.message.config.InfConfigRefreshMessage; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/infra/mq/message/config/InfConfigRefreshMessage.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/infra/mq/message/config/InfConfigRefreshMessage.java index a4f6e01..4330481 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/infra/mq/message/config/InfConfigRefreshMessage.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/infra/mq/message/config/InfConfigRefreshMessage.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.infra.mq.message.config; -import cn.iocoder.yudao.framework.redis.core.pubsub.ChannelMessage; +import cn.iocoder.yudao.framework.mq.core.pubsub.ChannelMessage; import lombok.Data; /** diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/infra/mq/producer/config/InfConfigProducer.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/infra/mq/producer/config/InfConfigProducer.java index 44d52e6..39a8e76 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/infra/mq/producer/config/InfConfigProducer.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/infra/mq/producer/config/InfConfigProducer.java @@ -1,7 +1,7 @@ package cn.iocoder.yudao.adminserver.modules.infra.mq.producer.config; -import cn.iocoder.yudao.framework.redis.core.util.RedisMessageUtils; import cn.iocoder.yudao.adminserver.modules.infra.mq.message.config.InfConfigRefreshMessage; +import cn.iocoder.yudao.framework.mq.core.util.RedisMessageUtils; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Component; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/dept/SysDeptRefreshConsumer.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/dept/SysDeptRefreshConsumer.java index 6673826..81bed50 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/dept/SysDeptRefreshConsumer.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/dept/SysDeptRefreshConsumer.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.consumer.dept; -import cn.iocoder.yudao.framework.redis.core.pubsub.AbstractChannelMessageListener; +import cn.iocoder.yudao.framework.mq.core.pubsub.AbstractChannelMessageListener; import cn.iocoder.yudao.adminserver.modules.system.mq.message.dept.SysDeptRefreshMessage; import cn.iocoder.yudao.adminserver.modules.system.service.dept.SysDeptService; import lombok.extern.slf4j.Slf4j; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/dict/SysDictDataRefreshConsumer.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/dict/SysDictDataRefreshConsumer.java index d7ff74d..08f4b81 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/dict/SysDictDataRefreshConsumer.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/dict/SysDictDataRefreshConsumer.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.consumer.dict; -import cn.iocoder.yudao.framework.redis.core.pubsub.AbstractChannelMessageListener; +import cn.iocoder.yudao.framework.mq.core.pubsub.AbstractChannelMessageListener; import cn.iocoder.yudao.adminserver.modules.system.mq.message.dict.SysDictDataRefreshMessage; import cn.iocoder.yudao.adminserver.modules.system.service.dict.SysDictDataService; import lombok.extern.slf4j.Slf4j; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/mail/SysMailSendConsumer.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/mail/SysMailSendConsumer.java index 3862987..ff3b017 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/mail/SysMailSendConsumer.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/mail/SysMailSendConsumer.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.consumer.mail; -import cn.iocoder.yudao.framework.redis.core.stream.AbstractStreamMessageListener; +import cn.iocoder.yudao.framework.mq.core.stream.AbstractStreamMessageListener; import cn.iocoder.yudao.adminserver.modules.system.mq.message.mail.SysMailSendMessage; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/permission/SysMenuRefreshConsumer.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/permission/SysMenuRefreshConsumer.java index 3d25416..9048e3c 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/permission/SysMenuRefreshConsumer.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/permission/SysMenuRefreshConsumer.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.consumer.permission; -import cn.iocoder.yudao.framework.redis.core.pubsub.AbstractChannelMessageListener; +import cn.iocoder.yudao.framework.mq.core.pubsub.AbstractChannelMessageListener; import cn.iocoder.yudao.adminserver.modules.system.mq.message.permission.SysMenuRefreshMessage; import cn.iocoder.yudao.adminserver.modules.system.service.permission.SysMenuService; import lombok.extern.slf4j.Slf4j; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/permission/SysRoleMenuRefreshConsumer.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/permission/SysRoleMenuRefreshConsumer.java index fc82706..5cdaeef 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/permission/SysRoleMenuRefreshConsumer.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/permission/SysRoleMenuRefreshConsumer.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.consumer.permission; -import cn.iocoder.yudao.framework.redis.core.pubsub.AbstractChannelMessageListener; +import cn.iocoder.yudao.framework.mq.core.pubsub.AbstractChannelMessageListener; import cn.iocoder.yudao.adminserver.modules.system.mq.message.permission.SysRoleMenuRefreshMessage; import cn.iocoder.yudao.adminserver.modules.system.service.permission.SysPermissionService; import lombok.extern.slf4j.Slf4j; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/permission/SysRoleRefreshConsumer.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/permission/SysRoleRefreshConsumer.java index 8bb2dee..a2f4cc5 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/permission/SysRoleRefreshConsumer.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/permission/SysRoleRefreshConsumer.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.consumer.permission; -import cn.iocoder.yudao.framework.redis.core.pubsub.AbstractChannelMessageListener; +import cn.iocoder.yudao.framework.mq.core.pubsub.AbstractChannelMessageListener; import cn.iocoder.yudao.adminserver.modules.system.mq.message.permission.SysRoleRefreshMessage; import cn.iocoder.yudao.adminserver.modules.system.service.permission.SysRoleService; import lombok.extern.slf4j.Slf4j; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/sms/SysSmsChannelRefreshConsumer.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/sms/SysSmsChannelRefreshConsumer.java index 21c1874..540ff17 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/sms/SysSmsChannelRefreshConsumer.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/sms/SysSmsChannelRefreshConsumer.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.consumer.sms; -import cn.iocoder.yudao.framework.redis.core.pubsub.AbstractChannelMessageListener; +import cn.iocoder.yudao.framework.mq.core.pubsub.AbstractChannelMessageListener; import cn.iocoder.yudao.adminserver.modules.system.mq.message.sms.SysSmsChannelRefreshMessage; import cn.iocoder.yudao.adminserver.modules.system.service.sms.SysSmsChannelService; import lombok.extern.slf4j.Slf4j; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/sms/SysSmsSendConsumer.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/sms/SysSmsSendConsumer.java index 68fa315..947b239 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/sms/SysSmsSendConsumer.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/sms/SysSmsSendConsumer.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.consumer.sms; -import cn.iocoder.yudao.framework.redis.core.stream.AbstractStreamMessageListener; +import cn.iocoder.yudao.framework.mq.core.stream.AbstractStreamMessageListener; import cn.iocoder.yudao.adminserver.modules.system.mq.message.sms.SysSmsSendMessage; import cn.iocoder.yudao.adminserver.modules.system.service.sms.SysSmsService; import lombok.extern.slf4j.Slf4j; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/sms/SysSmsTemplateRefreshConsumer.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/sms/SysSmsTemplateRefreshConsumer.java index b083243..1f4d92f 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/sms/SysSmsTemplateRefreshConsumer.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/consumer/sms/SysSmsTemplateRefreshConsumer.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.consumer.sms; -import cn.iocoder.yudao.framework.redis.core.pubsub.AbstractChannelMessageListener; +import cn.iocoder.yudao.framework.mq.core.pubsub.AbstractChannelMessageListener; import cn.iocoder.yudao.adminserver.modules.system.mq.message.sms.SysSmsTemplateRefreshMessage; import cn.iocoder.yudao.adminserver.modules.system.service.sms.SysSmsTemplateService; import lombok.extern.slf4j.Slf4j; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/dept/SysDeptRefreshMessage.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/dept/SysDeptRefreshMessage.java index 2aa5332..724547d 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/dept/SysDeptRefreshMessage.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/dept/SysDeptRefreshMessage.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.message.dept; -import cn.iocoder.yudao.framework.redis.core.pubsub.ChannelMessage; +import cn.iocoder.yudao.framework.mq.core.pubsub.ChannelMessage; import lombok.Data; /** diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/dict/SysDictDataRefreshMessage.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/dict/SysDictDataRefreshMessage.java index b4520fb..7b735de 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/dict/SysDictDataRefreshMessage.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/dict/SysDictDataRefreshMessage.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.message.dict; -import cn.iocoder.yudao.framework.redis.core.pubsub.ChannelMessage; +import cn.iocoder.yudao.framework.mq.core.pubsub.ChannelMessage; import lombok.Data; /** diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/mail/SysMailSendMessage.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/mail/SysMailSendMessage.java index cfea76b..bb9f621 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/mail/SysMailSendMessage.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/mail/SysMailSendMessage.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.message.mail; -import cn.iocoder.yudao.framework.redis.core.stream.StreamMessage; +import cn.iocoder.yudao.framework.mq.core.stream.StreamMessage; import lombok.Data; import javax.validation.constraints.NotNull; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/permission/SysMenuRefreshMessage.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/permission/SysMenuRefreshMessage.java index 73d611f..1fa2a38 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/permission/SysMenuRefreshMessage.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/permission/SysMenuRefreshMessage.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.message.permission; -import cn.iocoder.yudao.framework.redis.core.pubsub.ChannelMessage; +import cn.iocoder.yudao.framework.mq.core.pubsub.ChannelMessage; import lombok.Data; /** diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/permission/SysRoleMenuRefreshMessage.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/permission/SysRoleMenuRefreshMessage.java index 03e321c..8b9f50c 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/permission/SysRoleMenuRefreshMessage.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/permission/SysRoleMenuRefreshMessage.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.message.permission; -import cn.iocoder.yudao.framework.redis.core.pubsub.ChannelMessage; +import cn.iocoder.yudao.framework.mq.core.pubsub.ChannelMessage; import lombok.Data; /** diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/permission/SysRoleRefreshMessage.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/permission/SysRoleRefreshMessage.java index 9ae1d94..8d8d1e0 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/permission/SysRoleRefreshMessage.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/permission/SysRoleRefreshMessage.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.message.permission; -import cn.iocoder.yudao.framework.redis.core.pubsub.ChannelMessage; +import cn.iocoder.yudao.framework.mq.core.pubsub.ChannelMessage; import lombok.Data; /** diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/sms/SysSmsChannelRefreshMessage.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/sms/SysSmsChannelRefreshMessage.java index 7379188..a372956 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/sms/SysSmsChannelRefreshMessage.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/sms/SysSmsChannelRefreshMessage.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.message.sms; -import cn.iocoder.yudao.framework.redis.core.pubsub.ChannelMessage; +import cn.iocoder.yudao.framework.mq.core.pubsub.ChannelMessage; import lombok.Data; /** diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/sms/SysSmsSendMessage.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/sms/SysSmsSendMessage.java index f073640..34ebf91 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/sms/SysSmsSendMessage.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/sms/SysSmsSendMessage.java @@ -1,7 +1,7 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.message.sms; import cn.iocoder.yudao.framework.common.core.KeyValue; -import cn.iocoder.yudao.framework.redis.core.stream.StreamMessage; +import cn.iocoder.yudao.framework.mq.core.stream.StreamMessage; import lombok.Data; import javax.validation.constraints.NotNull; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/sms/SysSmsTemplateRefreshMessage.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/sms/SysSmsTemplateRefreshMessage.java index 9472f8b..c8bb00a 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/sms/SysSmsTemplateRefreshMessage.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/message/sms/SysSmsTemplateRefreshMessage.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.message.sms; -import cn.iocoder.yudao.framework.redis.core.pubsub.ChannelMessage; +import cn.iocoder.yudao.framework.mq.core.pubsub.ChannelMessage; import lombok.Data; /** diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/dept/SysDeptProducer.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/dept/SysDeptProducer.java index b4c974a..948796e 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/dept/SysDeptProducer.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/dept/SysDeptProducer.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.producer.dept; -import cn.iocoder.yudao.framework.redis.core.util.RedisMessageUtils; +import cn.iocoder.yudao.framework.mq.core.util.RedisMessageUtils; import cn.iocoder.yudao.adminserver.modules.system.mq.message.dept.SysDeptRefreshMessage; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Component; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/dict/SysDictDataProducer.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/dict/SysDictDataProducer.java index 5eb29eb..ea01837 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/dict/SysDictDataProducer.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/dict/SysDictDataProducer.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.producer.dict; -import cn.iocoder.yudao.framework.redis.core.util.RedisMessageUtils; +import cn.iocoder.yudao.framework.mq.core.util.RedisMessageUtils; import cn.iocoder.yudao.adminserver.modules.system.mq.message.dict.SysDictDataRefreshMessage; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Component; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/permission/SysMenuProducer.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/permission/SysMenuProducer.java index 4e1e6a6..6b34934 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/permission/SysMenuProducer.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/permission/SysMenuProducer.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.producer.permission; -import cn.iocoder.yudao.framework.redis.core.util.RedisMessageUtils; +import cn.iocoder.yudao.framework.mq.core.util.RedisMessageUtils; import cn.iocoder.yudao.adminserver.modules.system.mq.message.permission.SysMenuRefreshMessage; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Component; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/permission/SysPermissionProducer.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/permission/SysPermissionProducer.java index 310fefd..d9a1bfc 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/permission/SysPermissionProducer.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/permission/SysPermissionProducer.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.producer.permission; -import cn.iocoder.yudao.framework.redis.core.util.RedisMessageUtils; +import cn.iocoder.yudao.framework.mq.core.util.RedisMessageUtils; import cn.iocoder.yudao.adminserver.modules.system.mq.message.permission.SysRoleMenuRefreshMessage; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Component; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/permission/SysRoleProducer.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/permission/SysRoleProducer.java index 09579e7..6888f27 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/permission/SysRoleProducer.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/permission/SysRoleProducer.java @@ -1,6 +1,6 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.producer.permission; -import cn.iocoder.yudao.framework.redis.core.util.RedisMessageUtils; +import cn.iocoder.yudao.framework.mq.core.util.RedisMessageUtils; import cn.iocoder.yudao.adminserver.modules.system.mq.message.permission.SysRoleRefreshMessage; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Component; diff --git a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/sms/SysSmsProducer.java b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/sms/SysSmsProducer.java index 53aa973..8123628 100644 --- a/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/sms/SysSmsProducer.java +++ b/yudao-admin-server/src/main/java/cn/iocoder/yudao/adminserver/modules/system/mq/producer/sms/SysSmsProducer.java @@ -1,7 +1,7 @@ package cn.iocoder.yudao.adminserver.modules.system.mq.producer.sms; import cn.iocoder.yudao.framework.common.core.KeyValue; -import cn.iocoder.yudao.framework.redis.core.util.RedisMessageUtils; +import cn.iocoder.yudao.framework.mq.core.util.RedisMessageUtils; import cn.iocoder.yudao.adminserver.modules.system.mq.message.sms.SysSmsChannelRefreshMessage; import cn.iocoder.yudao.adminserver.modules.system.mq.message.sms.SysSmsSendMessage; import cn.iocoder.yudao.adminserver.modules.system.mq.message.sms.SysSmsTemplateRefreshMessage; diff --git a/yudao-admin-server/src/test-integration/java/cn/iocoder/yudao/adminserver/framework/redis/core/stream/RedisStreamTest.java b/yudao-admin-server/src/test-integration/java/cn/iocoder/yudao/adminserver/framework/redis/core/stream/RedisStreamTest.java index eea5c00..727b257 100644 --- a/yudao-admin-server/src/test-integration/java/cn/iocoder/yudao/adminserver/framework/redis/core/stream/RedisStreamTest.java +++ b/yudao-admin-server/src/test-integration/java/cn/iocoder/yudao/adminserver/framework/redis/core/stream/RedisStreamTest.java @@ -6,7 +6,7 @@ import cn.iocoder.yudao.adminserver.modules.system.mq.consumer.mail.SysMailSendC import cn.iocoder.yudao.adminserver.modules.system.mq.consumer.sms.SysSmsSendConsumer; import cn.iocoder.yudao.adminserver.modules.system.mq.message.mail.SysMailSendMessage; import cn.iocoder.yudao.adminserver.modules.system.mq.message.sms.SysSmsSendMessage; -import cn.iocoder.yudao.framework.redis.core.util.RedisMessageUtils; +import cn.iocoder.yudao.framework.mq.core.util.RedisMessageUtils; import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.springframework.context.annotation.Import; diff --git a/yudao-dependencies/pom.xml b/yudao-dependencies/pom.xml index 7522fff..7fab310 100644 --- a/yudao-dependencies/pom.xml +++ b/yudao-dependencies/pom.xml @@ -193,6 +193,13 @@ ${revision} + + + cn.iocoder.boot + yudao-spring-boot-starter-mq + ${revision} + + cn.iocoder.boot diff --git a/yudao-framework/pom.xml b/yudao-framework/pom.xml index 9363279..635dc7c 100644 --- a/yudao-framework/pom.xml +++ b/yudao-framework/pom.xml @@ -15,12 +15,16 @@ yudao-spring-boot-starter-redis yudao-spring-boot-starter-web yudao-spring-boot-starter-security + yudao-spring-boot-starter-monitor yudao-spring-boot-starter-protection yudao-spring-boot-starter-config yudao-spring-boot-starter-job + yudao-spring-boot-starter-mq + yudao-spring-boot-starter-excel yudao-spring-boot-starter-test + yudao-spring-boot-starter-biz-operatelog yudao-spring-boot-starter-biz-dict yudao-spring-boot-starter-biz-sms diff --git a/yudao-framework/yudao-spring-boot-starter-mq/pom.xml b/yudao-framework/yudao-spring-boot-starter-mq/pom.xml new file mode 100644 index 0000000..caef6f4 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/pom.xml @@ -0,0 +1,26 @@ + + + + cn.iocoder.boot + yudao-framework + ${revision} + + 4.0.0 + yudao-spring-boot-starter-mq + jar + + ${artifactId} + 消息队列，基于 Redis Pub/Sub 实现广播消费，基于 Stream 实现集群消费 + https://github.com/YunaiV/ruoyi-vue-pro + + + + + cn.iocoder.boot + yudao-spring-boot-starter-redis + + + + diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/config/YudaoMQAutoConfiguration.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/config/YudaoMQAutoConfiguration.java new file mode 100644 index 0000000..e105dd8 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/config/YudaoMQAutoConfiguration.java @@ -0,0 +1,105 @@ +package cn.iocoder.yudao.framework.mq.config; + +import cn.hutool.system.SystemUtil; +import cn.iocoder.yudao.framework.mq.core.pubsub.AbstractChannelMessageListener; +import cn.iocoder.yudao.framework.mq.core.stream.AbstractStreamMessageListener; +import lombok.extern.slf4j.Slf4j; +import org.springframework.boot.autoconfigure.AutoConfigureAfter; +import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration; +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.stream.StreamMessageListenerContainer; + +import java.util.List; + +/** + * 消息队列配置类 + * + * @author 芋道源码 + */ +@Configuration +@AutoConfigureAfter(YudaoMQAutoConfiguration.class) +@Slf4j +public class YudaoMQAutoConfiguration { + + /** + * 创建 Redis Pub/Sub 广播消费的容器 + */ + @Bean + public RedisMessageListenerContainer redisMessageListenerContainer( + RedisConnectionFactory factory, List> listeners) { + // 创建 RedisMessageListenerContainer 对象 + RedisMessageListenerContainer container = new RedisMessageListenerContainer(); + // 设置 RedisConnection 工厂。 + container.setConnectionFactory(factory); + // 添加监听器 + listeners.forEach(listener -> { + container.addMessageListener(listener, new ChannelTopic(listener.getChannel())); + log.info("[redisMessageListenerContainer][注册 Channel({}) 对应的监听器({})]", + listener.getChannel(), listener.getClass().getName()); + }); + 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; + } + + /** + * 构建消费者名字，使用本地 IP + 进程编号的方式。 + * 参考自 RocketMQ clientId 的实现 + * + * @return 消费者名字 + */ + private static String buildConsumerName() { + return String.format("%s@%d", SystemUtil.getHostInfo().getAddress(), SystemUtil.getCurrentPID()); + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/AbstractChannelMessageListener.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/AbstractChannelMessageListener.java new file mode 100644 index 0000000..9905a08 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/AbstractChannelMessageListener.java @@ -0,0 +1,71 @@ +package cn.iocoder.yudao.framework.mq.core.pubsub; + +import cn.hutool.core.util.TypeUtil; +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; +import lombok.SneakyThrows; +import org.springframework.data.redis.connection.Message; +import org.springframework.data.redis.connection.MessageListener; + +import java.lang.reflect.Type; + +/** + * Redis Pub/Sub 监听器抽象类，用于实现广播消费 + * + * @param 消息类型。一定要填写噢，不然会报错 + * + * @author 芋道源码 + */ +public abstract class AbstractChannelMessageListener implements MessageListener { + + /** + * 消息类型 + */ + private final Class messageType; + /** + * Redis Channel + */ + private final String channel; + + @SneakyThrows + protected AbstractChannelMessageListener() { + this.messageType = getMessageClass(); + this.channel = messageType.newInstance().getChannel(); + } + + /** + * 获得 Sub 订阅的 Redis Channel 通道 + * + * @return channel + */ + public final String getChannel() { + return channel; + } + + @Override + public final void onMessage(Message message, byte[] bytes) { + T messageObj = JsonUtils.parseObject(message.getBody(), messageType); + this.onMessage(messageObj); + } + + /** + * 处理消息 + * + * @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/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/ChannelMessage.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/ChannelMessage.java new file mode 100644 index 0000000..ff55f8b --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/ChannelMessage.java @@ -0,0 +1,20 @@ +package cn.iocoder.yudao.framework.mq.core.pubsub; + +import com.fasterxml.jackson.annotation.JsonIgnore; + +/** + * Redis Channel Message 接口 + * + * @author 芋道源码 + */ +public interface ChannelMessage { + + /** + * 获得 Redis Channel + * + * @return Channel + */ + @JsonIgnore // 避免序列化 + String getChannel(); + +} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/AbstractStreamMessageListener.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/AbstractStreamMessageListener.java new file mode 100644 index 0000000..612b5a0 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/AbstractStreamMessageListener.java @@ -0,0 +1,88 @@ +package cn.iocoder.yudao.framework.mq.core.stream; + +import cn.hutool.core.util.TypeUtil; +import cn.iocoder.yudao.framework.common.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/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/StreamMessage.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/StreamMessage.java new file mode 100644 index 0000000..30b38c6 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/StreamMessage.java @@ -0,0 +1,20 @@ +package cn.iocoder.yudao.framework.mq.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/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/util/RedisMessageUtils.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/util/RedisMessageUtils.java new file mode 100644 index 0000000..57c925f --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/util/RedisMessageUtils.java @@ -0,0 +1,40 @@ +package cn.iocoder.yudao.framework.mq.core.util; + +import cn.iocoder.yudao.framework.mq.core.pubsub.ChannelMessage; +import cn.iocoder.yudao.framework.mq.core.stream.StreamMessage; +import cn.iocoder.yudao.framework.common.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; + +/** + * Redis 消息工具类 + * + * @author 芋道源码 + */ +public class RedisMessageUtils { + + /** + * 发送 Redis 消息，基于 Redis pub/sub 实现 + * + * @param redisTemplate Redis 操作模板 + * @param 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/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/package-info.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/package-info.java new file mode 100644 index 0000000..48eaf23 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/package-info.java @@ -0,0 +1,6 @@ +/** + * 消息队列，基于 Redis 提供： + * 1. 基于 Pub/Sub 实现广播消费 + * 2. 基于 Stream 实现集群消费 + */ +package cn.iocoder.yudao.framework.mq; diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/resources/META-INF/spring.factories b/yudao-framework/yudao-spring-boot-starter-mq/src/main/resources/META-INF/spring.factories new file mode 100644 index 0000000..d4ca5b9 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ + cn.iocoder.yudao.framework.mq.config.YudaoMQAutoConfiguration diff --git a/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/config/YudaoRedisAutoConfiguration.java b/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/config/YudaoRedisAutoConfiguration.java index 4e9f0a0..8f4b5ad 100644 --- a/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/config/YudaoRedisAutoConfiguration.java +++ b/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/config/YudaoRedisAutoConfiguration.java @@ -1,23 +1,11 @@ package cn.iocoder.yudao.framework.redis.config; -import cn.hutool.system.SystemUtil; -import cn.iocoder.yudao.framework.redis.core.pubsub.AbstractChannelMessageListener; -import cn.iocoder.yudao.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; /** * Redis 配置类 @@ -44,76 +32,4 @@ public class YudaoRedisAutoConfiguration { return template; } - /** - * 创建 Redis Pub/Sub 广播消费的容器 - */ - @Bean - public RedisMessageListenerContainer redisMessageListenerContainer( - RedisConnectionFactory factory, List> listeners) { - // 创建 RedisMessageListenerContainer 对象 - RedisMessageListenerContainer container = new RedisMessageListenerContainer(); - // 设置 RedisConnection 工厂。 - container.setConnectionFactory(factory); - // 添加监听器 - listeners.forEach(listener -> { - container.addMessageListener(listener, new ChannelTopic(listener.getChannel())); - log.info("[redisMessageListenerContainer][注册 Channel({}) 对应的监听器({})]", - listener.getChannel(), listener.getClass().getName()); - }); - 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; - } - - /** - * 构建消费者名字，使用本地 IP + 进程编号的方式。 - * 参考自 RocketMQ clientId 的实现 - * - * @return 消费者名字 - */ - private static String buildConsumerName() { - return String.format("%s@%d", SystemUtil.getHostInfo().getAddress(), SystemUtil.getCurrentPID()); - } - } diff --git a/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/pubsub/AbstractChannelMessageListener.java b/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/pubsub/AbstractChannelMessageListener.java deleted file mode 100644 index 8d3bf6f..0000000 --- a/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/pubsub/AbstractChannelMessageListener.java +++ /dev/null @@ -1,71 +0,0 @@ -package cn.iocoder.yudao.framework.redis.core.pubsub; - -import cn.hutool.core.util.TypeUtil; -import cn.iocoder.yudao.framework.common.util.json.JsonUtils; -import lombok.SneakyThrows; -import org.springframework.data.redis.connection.Message; -import org.springframework.data.redis.connection.MessageListener; - -import java.lang.reflect.Type; - -/** - * Redis Pub/Sub 监听器抽象类，用于实现广播消费 - * - * @param 消息类型。一定要填写噢，不然会报错 - * - * @author 芋道源码 - */ -public abstract class AbstractChannelMessageListener implements MessageListener { - - /** - * 消息类型 - */ - private final Class messageType; - /** - * Redis Channel - */ - private final String channel; - - @SneakyThrows - protected AbstractChannelMessageListener() { - this.messageType = getMessageClass(); - this.channel = messageType.newInstance().getChannel(); - } - - /** - * 获得 Sub 订阅的 Redis Channel 通道 - * - * @return channel - */ - public final String getChannel() { - return channel; - } - - @Override - public final void onMessage(Message message, byte[] bytes) { - T messageObj = JsonUtils.parseObject(message.getBody(), messageType); - this.onMessage(messageObj); - } - - /** - * 处理消息 - * - * @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/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/pubsub/ChannelMessage.java b/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/pubsub/ChannelMessage.java deleted file mode 100644 index 60e5f49..0000000 --- a/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/pubsub/ChannelMessage.java +++ /dev/null @@ -1,20 +0,0 @@ -package cn.iocoder.yudao.framework.redis.core.pubsub; - -import com.fasterxml.jackson.annotation.JsonIgnore; - -/** - * Redis Channel Message 接口 - * - * @author 芋道源码 - */ -public interface ChannelMessage { - - /** - * 获得 Redis Channel - * - * @return Channel - */ - @JsonIgnore // 避免序列化 - String getChannel(); - -} diff --git a/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/stream/AbstractStreamMessageListener.java b/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/stream/AbstractStreamMessageListener.java deleted file mode 100644 index 85d44f4..0000000 --- a/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/stream/AbstractStreamMessageListener.java +++ /dev/null @@ -1,88 +0,0 @@ -package cn.iocoder.yudao.framework.redis.core.stream; - -import cn.hutool.core.util.TypeUtil; -import cn.iocoder.yudao.framework.common.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/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/stream/StreamMessage.java b/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/stream/StreamMessage.java deleted file mode 100644 index ecb9c74..0000000 --- a/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/stream/StreamMessage.java +++ /dev/null @@ -1,20 +0,0 @@ -package cn.iocoder.yudao.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/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/util/RedisMessageUtils.java b/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/util/RedisMessageUtils.java deleted file mode 100644 index 0d83586..0000000 --- a/yudao-framework/yudao-spring-boot-starter-redis/src/main/java/cn/iocoder/yudao/framework/redis/core/util/RedisMessageUtils.java +++ /dev/null @@ -1,40 +0,0 @@ -package cn.iocoder.yudao.framework.redis.core.util; - -import cn.iocoder.yudao.framework.redis.core.pubsub.ChannelMessage; -import cn.iocoder.yudao.framework.redis.core.stream.StreamMessage; -import cn.iocoder.yudao.framework.common.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; - -/** - * Redis 消息工具类 - * - * @author 芋道源码 - */ -public class RedisMessageUtils { - - /** - * 发送 Redis 消息，基于 Redis pub/sub 实现 - * - * @param redisTemplate Redis 操作模板 - * @param 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 - } - -} -- libgit2 0.21.4