1. 简单 redis stream 的 StreamMessage 和对应的消费者

2. 跑通 Redis Stream 的流程

1. 简单 redis stream 的 StreamMessage 和对应的消费者
2. 跑通 Redis Stream 的流程
YunaiV
1 parent f5331ce6
Showing 13 changed files with 458 additions and 3 deletions
src/main/java/cn/iocoder/dashboard/framework/redis/config/RedisConfig.java
src/main/java/cn/iocoder/dashboard/framework/redis/core/pubsub/ChannelMessage.java
src/main/java/cn/iocoder/dashboard/framework/redis/core/stream/AbstractStreamMessageListener.java
src/main/java/cn/iocoder/dashboard/framework/redis/core/stream/StreamMessage.java
src/main/java/cn/iocoder/dashboard/framework/redis/core/util/RedisMessageUtils.java
src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/mail/SysMailSendConsumer.java
src/main/java/cn/iocoder/dashboard/modules/system/mq/consumer/sms/SysSmsSendConsumer.java
src/main/java/cn/iocoder/dashboard/modules/system/mq/message/mail/SysMailSendMessage.java
src/main/java/cn/iocoder/dashboard/modules/system/mq/message/sms/SysSmsSendMessage.java
src/test-integration/java/cn/iocoder/dashboard/BaseRedisIntegrationTest.java
src/test-integration/java/cn/iocoder/dashboard/framework/redis/core/stream/RedisStreamTest.java
src/test-integration/resources/application-integration-test.yaml
src/test/java/cn/iocoder/dashboard/modules/system/service/dept/SysDeptServiceTest.java
 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<String, ObjectRecord<String, String>> redisStreamMessageListenerContainer(
+            RedisConnectionFactory factory, List<AbstractStreamMessageListener<?>> listeners) {
+        // 创建配置对象
+        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>>
+                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<String, ObjectRecord<String, String>> container = StreamMessageListenerContainer
+                .create(factory, streamMessageListenerContainerOptions);
+
+        RedisTemplate<String, Object> 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;
+    }
+
 }
@@ -4,6 +4,8 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
 /**
  * Redis Channel Message 接口
+ *
+ * @author 芋道源码
  */
 public interface ChannelMessage {
@@ -12,7 +14,7 @@ public interface ChannelMessage {
      *
      * @return Channel
      */
-    @JsonIgnore // 必须序列化
+    @JsonIgnore // 避免序列化
     String getChannel();
 }
+package cn.iocoder.dashboard.framework.redis.core.stream;
+
+import cn.hutool.core.util.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 <T> 消息类型。一定要填写噢，不然会报错
+ *
+ * @author 芋道源码
+ */
+public abstract class AbstractStreamMessageListener<T extends StreamMessage>
+        implements StreamListener<String, ObjectRecord<String, String>> {
+
+    /**
+     * 消息类型
+     */
+    private final Class<T> messageType;
+    /**
+     * Redis Channel
+     */
+    @Getter
+    private final String streamKey;
+
+    /**
+     * Redis 消费者分组，默认使用 spring.application.name 名字
+     */
+    @Value("spring.application.name")
+    @Getter
+    private String group;
+
+    @SneakyThrows
+    protected AbstractStreamMessageListener() {
+        this.messageType = getMessageClass();
+        this.streamKey = messageType.newInstance().getStreamKey();
+    }
+
+    @Override
+    public void onMessage(ObjectRecord<String, String> message) {
+        System.out.println(message);
+    }
+
+    /**
+     * 处理消息
+     *
+     * @param message 消息
+     */
+    public abstract void onMessage(T message);
+
+    /**
+     * 通过解析类上的泛型，获得消息类型
+     *
+     * @return 消息类型
+     */
+    @SuppressWarnings("unchecked")
+    // TODO 芋艿：稍后重构
+    private Class<T> 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<T>) types[0];
+        }
+        throw new IllegalStateException(String.format("类型(%s) 找不到 AbstractStreamMessageListener 父类", getClass().getName()));
+    }
+
+}
+package cn.iocoder.dashboard.framework.redis.core.stream;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+
+/**
+ * Redis Stream Message 接口
+ *
+ * @author 芋道源码
+ */
+public interface StreamMessage {
+
+    /**
+     * 获得 Redis Stream Key
+     *
+     * @return Channel
+     */
+    @JsonIgnore // 避免序列化
+    String getStreamKey();
+
+}
 package cn.iocoder.dashboard.framework.redis.core.util;
 import cn.iocoder.dashboard.framework.redis.core.pubsub.ChannelMessage;
+import cn.iocoder.dashboard.framework.redis.core.stream.StreamMessage;
 import cn.iocoder.dashboard.util.json.JsonUtils;
+import org.springframework.data.redis.connection.stream.RecordId;
+import org.springframework.data.redis.connection.stream.StreamRecords;
 import org.springframework.data.redis.core.RedisTemplate;
 /**
@@ -17,8 +20,21 @@ public class RedisMessageUtils {
      * @param redisTemplate Redis 操作模板
      * @param message 消息
      */
-    public static <T extends ChannelMessage>  void sendChannelMessage(RedisTemplate<?, ?> redisTemplate, T message) {
+    public static <T extends ChannelMessage> void sendChannelMessage(RedisTemplate<?, ?> redisTemplate, T message) {
         redisTemplate.convertAndSend(message.getChannel(), JsonUtils.toJsonString(message));
     }
+    /**
+     * 发送 Redis 消息，基于 Redis Stream 实现
+     *
+     * @param redisTemplate Redis 操作模板
+     * @param message 消息
+     * @return 消息记录的编号对象
+     */
+    public static <T extends StreamMessage> RecordId sendStreamMessage(RedisTemplate<String, String> redisTemplate, T message) {
+        return redisTemplate.opsForStream().add(StreamRecords.newRecord()
+                .ofObject(JsonUtils.toJsonString(message)) // 设置内容
+                .withStreamKey(message.getStreamKey())); // 设置 stream key
+    }
+
 }
+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<SysMailSendMessage> {
+
+    @Override
+    public void onMessage(SysMailSendMessage message) {
+
+    }
+
+}
+package cn.iocoder.dashboard.modules.system.mq.consumer.sms;
+
+import cn.iocoder.dashboard.framework.redis.core.stream.AbstractStreamMessageListener;
+import cn.iocoder.dashboard.modules.system.mq.message.sms.SysSmsSendMessage;
+import org.springframework.stereotype.Component;
+
+@Component
+public class SysSmsSendConsumer extends AbstractStreamMessageListener<SysSmsSendMessage> {
+
+    @Override
+    public void onMessage(SysSmsSendMessage message) {
+        System.out.println(message);
+    }
+
+}
+package cn.iocoder.dashboard.modules.system.mq.message.mail;
+
+import cn.iocoder.dashboard.framework.redis.core.stream.StreamMessage;
+import lombok.Data;
+
+import javax.validation.constraints.NotNull;
+import java.util.Map;
+
+/**
+ * 邮箱发送消息
+ *
+ * @author 芋道源码
+ */
+@Data
+public class SysMailSendMessage implements StreamMessage {
+
+    /**
+     * 邮箱地址
+     */
+    @NotNull(message = "邮箱地址不能为空")
+    private String address;
+    /**
+     * 短信模板编号
+     */
+    @NotNull(message = "短信模板编号不能为空")
+    private String templateCode;
+    /**
+     * 短信模板参数
+     */
+    private Map<String, Object> templateParams;
+
+    /**
+     * 用户编号，允许空
+     */
+    private Integer userId;
+    /**
+     * 用户类型，允许空
+     */
+    private Integer userType;
+
+    @Override
+    public String getStreamKey() {
+        return "system.mail.send";
+    }
+
+}
+package cn.iocoder.dashboard.modules.system.mq.message.sms;
+
+import cn.iocoder.dashboard.framework.redis.core.stream.StreamMessage;
+import lombok.Data;
+
+import javax.validation.constraints.NotNull;
+import java.util.Map;
+
+/**
+ * 短信发送消息
+ *
+ * @author 芋道源码
+ */
+@Data
+public class SysSmsSendMessage implements StreamMessage {
+
+    /**
+     * 手机号
+     */
+    @NotNull(message = "手机号不能为空")
+    private String mobile;
+    /**
+     * 短信模板编号
+     */
+    @NotNull(message = "短信模板编号不能为空")
+    private String templateCode;
+    /**
+     * 短信模板参数
+     */
+    private Map<String, Object> templateParams;
+
+    /**
+     * 用户编号，允许空
+     */
+    private Integer userId;
+    /**
+     * 用户类型，允许空
+     */
+    private Integer userType;
+
+    @Override
+    public String getStreamKey() {
+        return "system.sms.send";
+    }
+
+}
+package cn.iocoder.dashboard;
+
+import cn.iocoder.dashboard.framework.redis.config.RedisConfig;
+import org.redisson.spring.starter.RedissonAutoConfiguration;
+import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Import;
+import org.springframework.test.context.ActiveProfiles;
+
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, classes = BaseRedisIntegrationTest.Application.class)
+@ActiveProfiles("integration-test") // 设置使用 application-integration-test 配置文件
+public class BaseRedisIntegrationTest {
+
+    @Import({
+            // Redis 配置类
+            RedisAutoConfiguration.class, // Spring Redis 自动配置类
+            RedisConfig.class, // 自己的 Redis 配置类
+            RedissonAutoConfiguration.class, // Redisson 自动高配置类
+    })
+    public static class Application {
+    }
+
+}
+package cn.iocoder.dashboard.framework.redis.core.stream;
+
+import cn.hutool.core.thread.ThreadUtil;
+import cn.iocoder.dashboard.BaseRedisIntegrationTest;
+import cn.iocoder.dashboard.framework.redis.core.util.RedisMessageUtils;
+import cn.iocoder.dashboard.modules.system.mq.consumer.mail.SysMailSendConsumer;
+import cn.iocoder.dashboard.modules.system.mq.consumer.sms.SysSmsSendConsumer;
+import cn.iocoder.dashboard.modules.system.mq.message.mail.SysMailSendMessage;
+import cn.iocoder.dashboard.modules.system.mq.message.sms.SysSmsSendMessage;
+import org.junit.jupiter.api.Test;
+import org.springframework.context.annotation.Import;
+import org.springframework.data.redis.core.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);
+        }
+
+    }
+
+}
+spring:
+  main:
+    lazy-initialization: true # 开启懒加载，加快速度
+    banner-mode: off # 单元测试，禁用 Banner
+
+--- #################### 数据库相关配置 ####################
+
+spring:
+  # 数据源配置项
+  datasource:
+    name: ruoyi-vue-pro
+    url: jdbc:h2:mem:testdb;MODE=MYSQL;DATABASE_TO_UPPER=false; # MODE 使用 MySQL 模式；DATABASE_TO_UPPER 配置表和字段使用小写
+    driver-class-name: org.h2.Driver
+    username: sa
+    password:
+    schema: classpath:sql/create_tables.sql # MySQL 转 H2 的语句，使用 https://www.jooq.org/translate/ 工具
+    druid:
+      async-init: true # 单元测试，异步初始化 Druid 连接池，提升启动速度
+      initial-size: 1 # 单元测试，配置为 1，提升启动速度
+
+  # Redis 配置。Redisson 默认的配置足够使用，一般不需要进行调优
+  redis:
+    host: 127.0.0.1 # 地址
+    port: 6379 # 端口（单元测试，使用 16379 端口）
+    database: 0 # 数据库索引
+
+mybatis:
+  lazy-initialization: true # 单元测试，设置 MyBatis Mapper 延迟加载，加速每个单元测试
+
+--- #################### 定时任务相关配置 ####################
+
+--- #################### 配置中心相关配置 ####################
+
+--- #################### 服务保障相关配置 ####################
+
+# Lock4j 配置项（单元测试，禁用 Lock4j）
+
+# Resilience4j 配置项
+resilience4j:
+  ratelimiter:
+    instances:
+      backendA:
+        limit-for-period: 1 # 每个周期内，允许的请求数。默认为 50
+        limit-refresh-period: 60s # 每个周期的时长，单位：微秒。默认为 500
+        timeout-duration: 1s # 被限流时，阻塞等待的时长，单位：微秒。默认为 5s
+        register-health-indicator: true # 是否注册到健康监测
+
+--- #################### 监控相关配置 ####################
+
+--- #################### 芋道相关配置 ####################
+
+# 芋道配置项，设置当前项目所有自定义的配置
+yudao:
+  info:
+    version: 1.0.0
+    base-package: cn.iocoder.dashboard
+  web:
+    api-prefix: /api
+    controller-package: ${yudao.info.base-package}
+  security:
+    token-header: Authorization
+    token-secret: abcdefghijklmnopqrstuvwxyz
+    token-timeout: 1d
+    session-timeout: 30m
+    mock-enable: true
+    mock-secret: test
+  swagger:
+    enable: false # 单元测试，禁用 Swagger
+  captcha:
+    timeout: 5m
+    width: 160
+    height: 60
+  file:
+    base-path: http://127.0.0.1:${server.port}/${yudao.web.api-prefix}/file/get/
+  codegen:
+    base-package: ${yudao.info.base-package}.modules
+    db-schemas: ${spring.datasource.name}
+  xss:
+    enable: false
+    exclude-urls: # 如下两个 url，仅仅是为了演示，去掉配置也没关系
+      - ${spring.boot.admin.context-path}/** # 不处理 Spring Boot Admin 的请求
+      - ${management.endpoints.web.base-path}/** # 不处理 Actuator 的请求
@@ -72,7 +72,6 @@ class SysDeptServiceTest extends BaseDbUnitTest {
         // 断言 maxUpdateTime 缓存
         Date maxUpdateTime = (Date) getFieldValue(deptService, "maxUpdateTime");
         assertEquals(ObjectUtils.max(deptDO1.getUpdateTime(), deptDO2.getUpdateTime()), maxUpdateTime);
-
     }
     @Test