1 前言

Redis 基本上是互聯(lián)網(wǎng)公司必備的工具了，Redis的應(yīng)用場景實在太多了，但是有很多相似的功能如果每個項目都要實現(xiàn)一遍就顯得太麻煩了，所以為了方便，我打算開發(fā)一個基于 Redis 的工具集，盡量做到開箱即用。

2 目前實現(xiàn)功能

這個工具集并沒有開發(fā)完成，實現(xiàn)了部分功能，如下圖

簡單介紹下已經(jīng)實現(xiàn)的模塊：

common : 整個項目公共模塊，比如AOP工具等；

delay: Redis實現(xiàn)的延遲隊列；

lock: Redis實現(xiàn)的分布式鎖；

mq: Redis實現(xiàn)消息隊列；

query: Redis實現(xiàn)分頁模糊查詢；

web: Redis實現(xiàn)web相關(guān)的功能；

duplicate ：防止重復(fù)提交；

以上的這些模塊都是已經(jīng)實現(xiàn)的了，還有 社交、限流、冪等相關(guān)功能后面會陸續(xù)實現(xiàn)。

3 如何使用

1.引入 Maven 依賴

目前可以下載代碼上傳到自己的私服或者本地倉庫，后面會推到 Maven 中央倉庫

cn.org.wangchangjiu
redis-util-spring-boot-starter
1.0.0-SNAPSHOT

2.配置文件（application.yaml）開啟各模塊功能開關(guān)

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

3.實現(xiàn)消息發(fā)送者

MQ消息發(fā)送：

延遲消息發(fā)送：

4.實現(xiàn)消息監(jiān)聽器

MQ消息監(jiān)聽器：

延遲消息監(jiān)聽器：

4 MQ和delay實現(xiàn)細節(jié)

MQ實現(xiàn)細節(jié)

容器啟動時，簡單來說就是通過springboot自動裝配，創(chuàng)建一些Bean，如下圖：

值得注意的是，springboot3.X 自動裝配方式有點變化，需要創(chuàng)建文件 META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports 文件，文件內(nèi)容就直接寫 自動配置類

RedisUtilAutoConfiguration 主自動裝配類會 import 各個模塊的自動裝配類：

我們以 RedisStreamAutoConfiguration 為例：

該裝配類生效需要顯示打開，然后就是創(chuàng)建各種Bean。

最主要的Bean有：

RedisMessageConsumerManager：

該Bean實現(xiàn)了 BeanPostProcessor 接口，主要作用是，獲取被注解 RedisMessageListener 修飾的方法，把信息封裝在 RedisMessageConsumerContainer 對象里，方便后面反射調(diào)用。

StreamMessageListenerContainer：

這個Bean主要是做 redis MQ 的配置，比如配置：一次最多獲取多少條消息、沒有消息時阻塞時間、執(zhí)行任務(wù)的executor、錯誤處理器、以及消費組、是否自動ACK等配置，具體代碼如下：

@Bean(initMethod="start",destroyMethod="stop")
@DependsOn("redisMessageConsumerManager")
@ConditionalOnMissingBean
publicStreamMessageListenerContainer>streamMessageListenerContainer(@AutowiredRedisMessageConsumerManagerredisMessageConsumerManager,
@AutowiredRedisConnectionFactoryredisConnectionFactory,
@AutowiredErrorHandlererrorHandler){
MyRedisStreamProperties.Optionsoptions=myRedisStreamProperties.getOptions();
StreamMessageListenerContainer.StreamMessageListenerContainerOptions>containerOptions=
StreamMessageListenerContainer.StreamMessageListenerContainerOptions
.builder()
//一次最多獲取多少條消息
.batchSize(options.getBatchSize())
//運行Stream的polltask
.executor(getStreamMessageListenerExecutor())
//Stream中沒有消息時，阻塞多長時間，需要比`spring.redis.timeout`的時間小
.pollTimeout(options.getPollTimeout())
//獲取消息的過程或獲取到消息給具體的消息者處理的過程中，發(fā)生了異常的處理
.errorHandler(errorHandler)
.build();

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

//獲取被RedisMessageListener注解修飾的bean
MapconsumerContainerGroups=
redisMessageConsumerManager.getConsumerContainerGroups();

//循環(huán)遍歷，創(chuàng)建消費組
consumerContainerGroups.forEach((groupQueue,redisMessageConsumerContainer)->{
String[]groupQueues=groupQueue.split("#");

//創(chuàng)建消費組
createGroups(groupQueues);

RedisMessageListenerredisMessageListener=redisMessageConsumerContainer.getRedisMessageListener();
if(!redisMessageListener.useGroup()){
//獨立消費不使用組
streamMessageListenerContainer.receive(StreamOffset.fromStart(groupQueues[1]),newDefaultGroupStreamListener(redisMessageConsumerContainer));
}else{
//消費組消費
if(redisMessageListener.autoAck()){
//自動ACK
streamMessageListenerContainer.receiveAutoAck(Consumer.from(groupQueues[0],"consumer:"+UUID.randomUUID()),
StreamOffset.create(groupQueues[1],ReadOffset.lastConsumed()),newDefaultGroupStreamListener(redisMessageConsumerContainer));
}else{
//手動ACK
streamMessageListenerContainer.receive(Consumer.from(groupQueues[0],"consumer:"+UUID.randomUUID()),
StreamOffset.create(groupQueues[1],ReadOffset.lastConsumed()),newDefaultGroupStreamListener(redisMessageConsumerContainer));
}
}
});
returnstreamMessageListenerContainer;
}

/**
*創(chuàng)建消費組
*@paramgroupQueues
*/
privatevoidcreateGroups(String[]groupQueues){
//判斷是否存在隊列Key
if(stringRedisTemplate.hasKey(groupQueues[1])){
//獲取消費組沒有則創(chuàng)建
StreamInfo.XInfoGroupsgroups=stringRedisTemplate.opsForStream().groups(groupQueues[1]);
if(groups.isEmpty()){
stringRedisTemplate.opsForStream().createGroup(groupQueues[1],groupQueues[0]);
}else{
AtomicBooleanexists=newAtomicBoolean(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后面這個線程池也可以交由用戶配置
privateExecutorgetStreamMessageListenerExecutor(){
AtomicIntegerindex=newAtomicInteger(1);
intprocessors=Runtime.getRuntime().availableProcessors();
ThreadPoolExecutorexecutor=newThreadPoolExecutor(processors,processors,0,TimeUnit.SECONDS,
newLinkedBlockingDeque<>(),r->{
Threadthread=newThread(r);
thread.setName("async-stream-consumer-"+index.getAndIncrement());
thread.setDaemon(true);
returnthread;
});
returnexecutor;
}

發(fā)送消息流程：

redis 延遲隊列的實現(xiàn)原理和這個差不多，主要是 redission延遲隊列 + 自定義注解 + 反射，代碼都差不多。