⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处 juejin.cn/post/7254574621898686521 「编程小学生哈」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表

RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址

您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。

新的源码解析文章实时收到通知。每周更新一篇左右。

认真的源码交流微信群。

1 前言

Redis 基本上是互联网公司必备的工具了，Redis的应用场景实在太多了，但是有很多相似的功能如果每个项目都要实现一遍就显得太麻烦了，所以为了方便，我打算开发一个基于 Redis 的工具集，尽量做到开箱即用。

2 目前实现功能

这个工具集并没有开发完成，实现了部分功能，如下图

简单介绍下已经实现的模块：

common : 整个项目公共模块，比如AOP工具等；
delay: Redis实现的延迟队列；
lock: Redis实现的分布式锁；
mq: Redis实现消息队列；
query: Redis实现分页模糊查询；
web: Redis实现web相关的功能；
- duplicate ：防止重复提交；

以上的这些模块都是已经实现的了，还有社交、限流、幂等相关功能后面会陆续实现。

3 如何使用

1.引入 Maven 依赖

目前可以下载代码上传到自己的私服或者本地仓库，后面会推到 Maven 中央仓库

<dependency>
    <groupId>cn.org.wangchangjiu</groupId>
    <artifactId>redis-util-spring-boot-starter</artifactId>
    <version>1.0.0-SNAPSHOT</version>
</dependency>

2.配置文件（application.yaml）开启各模块功能开关

redis:
  util:
    mq:
      enable: true
    delay:
      enable: true

3.实现消息发送者

MQ消息发送：

延迟消息发送：

4.实现消息监听器

MQ消息监听器：

延迟消息监听器：

4 MQ和delay实现细节

MQ实现细节

容器启动时，简单来说就是通过springboot自动装配，创建一些Bean，如下图：

值得注意的是，springboot3.X 自动装配方式有点变化，需要创建文件 META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports 文件，文件内容就直接写自动配置类

RedisUtilAutoConfiguration 主自动装配类会 import 各个模块的自动装配类：

我们以 RedisStreamAutoConfiguration 为例：

该装配类生效需要显示打开，然后就是创建各种Bean。

最主要的Bean有：

RedisMessageConsumerManager：

该Bean实现了 BeanPostProcessor 接口，主要作用是，获取被注解 RedisMessageListener 修饰的方法，把信息封装在 RedisMessageConsumerContainer 对象里，方便后面反射调用。

StreamMessageListenerContainer：

这个Bean主要是做 redis MQ 的配置，比如配置：一次最多获取多少条消息、没有消息时阻塞时间、执行任务的executor、错误处理器、以及消费组、是否自动ACK等配置，具体代码如下：

@Bean(initMethod = "start", destroyMethod = "stop")
@DependsOn("redisMessageConsumerManager")
@ConditionalOnMissingBean
public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(@Autowired RedisMessageConsumerManager redisMessageConsumerManager,
                                                                                                                @Autowired RedisConnectionFactory redisConnectionFactory,
                                                                                                                @Autowired ErrorHandler errorHandler) {
    MyRedisStreamProperties.Options options = myRedisStreamProperties.getOptions();
    StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> containerOptions =
            StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                    .builder()
                    // 一次最多获取多少条消息
                    .batchSize(options.getBatchSize())
                    // 运行 Stream 的 poll task
                    .executor(getStreamMessageListenerExecutor())
                    // Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小
                    .pollTimeout(options.getPollTimeout())
                    // 获取消息的过程或获取到消息给具体的消息者处理的过程中，发生了异常的处理
                    .errorHandler(errorHandler)
                    .build();

    StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer =
            StreamMessageListenerContainer.create(redisConnectionFactory, containerOptions);

    // 获取 被 RedisMessageListener 注解修饰的 bean
    Map<String, RedisMessageConsumerContainer> consumerContainerGroups =
            redisMessageConsumerManager.getConsumerContainerGroups();

    // 循环遍历，创建 消费组
    consumerContainerGroups.forEach((groupQueue, redisMessageConsumerContainer) -> {
        String[] groupQueues = groupQueue.split("#");

        // 创建消费组
        createGroups(groupQueues);

        RedisMessageListener redisMessageListener = redisMessageConsumerContainer.getRedisMessageListener();
        if(!redisMessageListener.useGroup()){
            // 独立消费 不使用组
            streamMessageListenerContainer.receive(StreamOffset.fromStart(groupQueues[1]), new DefaultGroupStreamListener(redisMessageConsumerContainer));
        } else {
            // 消费组 消费
            if(redisMessageListener.autoAck()){
                // 自动ACK
                streamMessageListenerContainer.receiveAutoAck(Consumer.from(groupQueues[0], "consumer:" + UUID.randomUUID()),
                        StreamOffset.create(groupQueues[1], ReadOffset.lastConsumed()), new DefaultGroupStreamListener(redisMessageConsumerContainer));
            } else {
                // 手动 ACK
                streamMessageListenerContainer.receive(Consumer.from(groupQueues[0], "consumer:" + UUID.randomUUID()),
                        StreamOffset.create(groupQueues[1], ReadOffset.lastConsumed()), new DefaultGroupStreamListener(redisMessageConsumerContainer));
            }
        }
    });
    return streamMessageListenerContainer;
}

/**
 *  创建消费组
 * @param groupQueues
 */
private void createGroups(String[] groupQueues) {
    // 判断是否存在队列Key
    if (stringRedisTemplate.hasKey(groupQueues[1])) {
        // 获取消费组 没有则创建
        StreamInfo.XInfoGroups groups = stringRedisTemplate.opsForStream().groups(groupQueues[1]);
        if (groups.isEmpty()) {
            stringRedisTemplate.opsForStream().createGroup(groupQueues[1], groupQueues[0]);
        } else {
            AtomicBoolean exists= new AtomicBoolean(false);
            groups.forEach(xInfoGroup -> {
                if (xInfoGroup.groupName().equals(groupQueues[0])){
                    exists.set(true);
                }
            });
            if(!exists.get()){
                stringRedisTemplate.opsForStream().createGroup(groupQueues[1], groupQueues[0]);
            }
        }
    } else {
        stringRedisTemplate.opsForStream().createGroup(groupQueues[1], groupQueues[0]);
    }
}

// todo 后面这个线程池也可以交由用户配置
private Executor getStreamMessageListenerExecutor() {
    AtomicInteger index = new AtomicInteger(1);
    int processors = Runtime.getRuntime().availableProcessors();
    ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,
            new LinkedBlockingDeque<>(), r -> {
        Thread thread = new Thread(r);
        thread.setName("async-stream-consumer-" + index.getAndIncrement());
        thread.setDaemon(true);
        return thread;
    });
    return executor;
}

发送消息流程：

redis 延迟队列的实现原理和这个差不多，主要是 redission延迟队列 + 自定义注解 + 反射，代码都差不多。

5 工具地址

https://gitee.com/listen_w/redis-util-spring-boot-starter

https://github.com/jettwangcj/redis-util-spring-boot-starter

芋道源码 —— 纯源码解析博客

为了方便开发，我打算实现一个Redis 工具集