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

系统管理员 2023-10-13 16:10 40阅读 0赞

#### 前言 ####

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

#### 目前实现功能 ####

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

![format_png][]

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

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

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

#### 如何使用 ####

1.  引入 Maven 依赖（目前可以下载代码上传到自己的私服或者本地仓库，后面会推到 Maven 中央仓库）
    
    xml
    
    复制代码
    
    `<dependency> <groupId>cn.org.wangchangjiu</groupId> <artifactId>redis-util-spring-boot-starter</artifactId> <version>1.0.0-SNAPSHOT</version> </dependency>`
2.  配置文件（application.yaml）开启各模块功能开关
    
    yaml
    
    复制代码
    
    `redis: util: mq: enable: true delay: enable: true`
3.  实现消息发送者
    
     *  MQ消息发送：
        
        ![format_png 1][]
     *  延迟消息发送：
        
        ![format_png 2][]
4.  实现消息监听器
    
     *  MQ消息监听器：
        
        ![format_png 3][]
     *  延迟消息监听器：
        
        ![format_png 4][]

#### MQ和delay实现细节 ####

##### MQ实现细节 #####

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

![format_png 5][]

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

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

![format_png 7][]

我们以 RedisStreamAutoConfiguration 为例：

![format_png 8][]

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

最主要的Bean有：

*  **RedisMessageConsumerManager**：
    
    ![format_png 9][]
    
    该Bean实现了 BeanPostProcessor 接口，主要作用是，获取被注解 RedisMessageListener 修饰的方法，把信息封装在 RedisMessageConsumerContainer 对象里，方便后面反射调用。
 *  StreamMessageListenerContainer：  
    这个Bean主要是做 redis MQ 的配置，比如配置：一次最多获取多少条消息、没有消息时阻塞时间、执行任务的executor、错误处理器、以及消费组、是否自动ACK等配置，具体代码如下：

less

复制代码

``@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; }``

发送消息流程：

![format_png 10][]

redis 延迟队列的实现原理和这个差不多，主要是 redission延迟队列 + 自定义注解 + 反射，代码都差不多，我之前写过一篇基于 redission + kafka的延迟队列实现方式类似，只是把kafka那部分除去了，具体可以看那篇博文，地址： [基于 Redisson 和 Kafka 的延迟队列设计方案][Redisson _ Kafka]

### 后记 ###

其他模块的设计细节后面再说，欢迎大家使用，也请大家多多提建议以及好的功能点，我都可以整合上去。

[format_png]: https://img-blog.csdnimg.cn/img_convert/0300c66f8f0a4d670e4468622266ad83.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/6373f08da477b00abd1719c276fc2511.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/f28c7301737c072b5426298201449e85.webp?x-oss-process=image/format,png
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/80595c9e2b0cabbea92c06cdb28958b5.webp?x-oss-process=image/format,png
[format_png 4]: https://img-blog.csdnimg.cn/img_convert/bd7d34d5a27aea3e9a35140b95d1be44.webp?x-oss-process=image/format,png
[format_png 5]: https://img-blog.csdnimg.cn/img_convert/f41d784df7885c199e09e9792685296e.webp?x-oss-process=image/format,png
[format_png 6]: https://img-blog.csdnimg.cn/img_convert/7ae9f96ed1ca53a01776fc72d8901d26.webp?x-oss-process=image/format,png
[format_png 7]: https://img-blog.csdnimg.cn/img_convert/f5dc0e2b41255fc27865799a206363c1.webp?x-oss-process=image/format,png
[format_png 8]: https://img-blog.csdnimg.cn/img_convert/06745222c400743271455677829e2afa.webp?x-oss-process=image/format,png
[format_png 9]: https://img-blog.csdnimg.cn/img_convert/6b72f5d1cc77c2c63b082a92067c8253.webp?x-oss-process=image/format,png
[format_png 10]: https://img-blog.csdnimg.cn/img_convert/eee4089b63cbefbafc3375b20c52bbc0.webp?x-oss-process=image/format,png
[Redisson _ Kafka]: https://juejin.cn/post/7144969196542099469