前言
上周发生了一件鸡飞狗跳的线上事故，六节点的Redis-Cluster集群所在的大部分机器因为网络带宽问题断连了，排查之后发现是那几台物理机带宽被占满了，导致整个集群因为槽位不满16384而请求失败。并且因为没有考虑缓存失效问题，而让有使用缓存的接口全部报错，影响了用户体验。在本地测试时，还发现因为框架中的Redisson组件初始化时会强制连接和校验槽位，而导致整个服务因Redis-Cluster断连而无法启动。
造成了一段时间的中断，当时是别的同事在负责，因为涉及到我的六脉神剑，所以我决定还是我来收个尾。经过上周五和本周一的方案思考和设计，琢磨出了一套不成熟的方案，暂行，期待读者有更好的想法能分享一下。同时我也会逐步分析我的思考过程，和最终方案设计时的考量。
现状导向问题
一开始上手比较匆忙，毕竟这个问题我一开始是抱着做甩手掌柜的心态来做旁观者。(⊙﹏⊙)，有个三四天吧，可能最后还是没有个好的解法，波及到我负责的项目了，想了想，还是从框架层控制一下，做一下整体规划。
简单描述下现状，一是Redis集群不可用了，用到Redis的接口就挂了，只有少部分接口做了手动降级处理，即捕获异常走数据库。二是本地想连这个集群，发现服务根本无法启动。根据现状整理问题，得出需要解决的是集群宕机后，如何保证服务正常运行以及正常启动。
思维风暴
代码规范
框架中作为Redis的入口和客户端，提供了三种方式，分别是SpringCache、SpringDataRedis和Redisson。同事在使用的时候也是随便用，因此我针对这三种都进行了一定的配置，详情见六脉神剑-我在公司造了六个轮子(21745阅读245赞536收藏)。使用情况比较复杂的时候，就要考虑周全，避免牵一发而动全身。
从问题一来看就是一个很明显的缓存失效场景，最直接的解决方案就是捕获异常后转数据库连接，但这是开发注意事项，是最直白的规范，详见三个月前我的一篇文章，同事血压操作集锦第一弹，截图如下。
但是今天要讨论的是要从框架层面解决这个问题，而且是要尽可能小的改动。因为老项目很多不遵守开发规范的代码，所以如果要统一按照上面的案例进行修改，那将是一个非常耗时的工程。同时还要解决第二个问题，Redis集群宕机后项目无法启动，因为框架强依赖于Redis，Spring会因为无法正常加载Redis相关的Bean而导致启动失败。
统一入口，减少变量
在进行了小半天的头脑风暴和资料查询后，发现大家没有这样的困扰，并且所有的方案都指向了Redis的高可用和代码规范。诚然，这是最优解，但是基于团队目前的现状，我也想做出我的思考，给出一套应急的方案。
说干就干，我的第一想法是统一入口，因为公司框架的SpringBoot版本是2.1.X过低，所以我个人极力不推荐使用SpringDataRedis，也就是redisTemplate。因此我干掉了这个依赖，只留下了原生的redisson作为唯一的Redis客户端。因为对SpringCache做了Redis的扩展，所以也更换成了Redisson提供支持(最开始用的lettuce，因为有问题就换成了Jedis，更换的原因见六脉神剑那篇文章)。
/**
* 定义Jedis客户端，集群和单点同时存在时优先集群配置
*/
@Bean
public JedisConnectionFactory redisConnectionFactory() {
String redisHost = config.getProperty("spring.redis.host");
String redisPort = config.getProperty("spring.redis.port");
String cluster = config.getProperty("spring.redis.cluster.nodes");
String redisPassword = config.getProperty("spring.redis.password");
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
// 默认阻塞等待时间为无限长，源码DEFAULT_MAX_WAIT_MILLIS = -1L
// 最大连接数, 根据业务需要设置，不能超过实例规格规定的最大连接数。
jedisPoolConfig.setMaxTotal(100);
// 最大空闲连接数, 根据业务需要设置，不能超过实例规格规定的最大连接数。
jedisPoolConfig.setMaxIdle(60);
// 关闭 testOn[Borrow|Return]，防止产生额外的PING。
jedisPoolConfig.setTestOnBorrow(false);
jedisPoolConfig.setTestOnReturn(false);
JedisClientConfiguration jedisClientConfiguration = JedisClientConfiguration.builder().usePooling()
.poolConfig(jedisPoolConfig).build();
if (StringUtils.hasText(cluster)) {
// 集群模式
String[] split = cluster.split(",");
RedisClusterConfiguration clusterServers = new RedisClusterConfiguration(Arrays.asList(split));
if (StringUtils.hasText(redisPassword)) {
clusterServers.setPassword(redisPassword);
}
return new JedisConnectionFactory(clusterServers, jedisClientConfiguration);
} else if (StringUtils.hasText(redisHost) && StringUtils.hasText(redisPort)) {
// 单机模式
RedisStandaloneConfiguration singleServer = new RedisStandaloneConfiguration(redisHost, Integer.parseInt(redisPort));
if (StringUtils.hasText(redisPassword)) {
singleServer.setPassword(redisPassword);
}
return new JedisConnectionFactory(singleServer, jedisClientConfiguration);
} else {
throw new ToolException("spring.redis.host及port或spring.redis.cluster" +
".nodes必填，否则不可使用RedisTool以及Redisson");
}
}
/**
* 配置Spring-Cache内部使用Redis，配置序列化和过期时间
*/
@Bean
public RedisCacheManager cacheManager(RedisConnectionFactory connectionFactory) {
RedisSerializer redisSerializer = new StringRedisSerializer();
Jackson2JsonRedisSerializer jackson2JsonRedisSerializer
= new Jackson2JsonRedisSerializer(Object.class);
ObjectMapper om = new ObjectMapper();
// 防止在序列化的过程中丢失对象的属性
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
// 开启实体类和json的类型转换，该处兼容老版本依赖，不得修改
om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jackson2JsonRedisSerializer.setObjectMapper(om);
// 配置序列化（解决乱码的问题）
RedisCacheConfiguration config = RedisCacheConfiguration.
defaultCacheConfig()
.serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(redisSerializer))
.serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer))
.disableCachingNullValues()// 不缓存空值
.entryTtl(Duration.ofMinutes(30));//30分钟不过期
return RedisCacheManager
.builder(connectionFactory)
.cacheDefaults(config)
.build();
}
调整后代码，就是按照官网改了改，感兴趣可以看看Redisson-WIKI
/**
* 对 Redisson 的使用都是通过 RedissonClient 对象
*/
@ConditionalOnProperty(name = "ruijie.tool.redis-enable", havingValue = "true")
@Bean(destroyMethod = "shutdown") // 服务停止后调用 shutdown 方法。
public RedissonClient redisson() {
String redisHost = config.getProperty("spring.redis.host");
String redisPort = config.getProperty("spring.redis.port");
String cluster = config.getProperty("spring.redis.cluster.nodes");
String redisPassword = config.getProperty("spring.redis.password");
Config config = new Config();
//使用String序列化时会出现RBucket转换异常
//config.setCodec(new StringCodec());
if (ObjectUtils.isEmpty(redisHost) && ObjectUtils.isEmpty(cluster)) {
throw new ToolException("spring.redis.host及port或spring.redis.cluster" +
".nodes必填，否则不可使用Redis");
} else {
if (StringUtils.hasText(cluster)) {
// 集群模式
String[] split = cluster.split(",");
List servers = new ArrayList();
for (String s : split) {
servers.add("redis://" + s);
}
ClusterServersConfig clusterServers = config.useClusterServers();
clusterServers.addNodeAddress(servers.toArray(new String[split.length]));
if (StringUtils.hasText(redisPassword)) {
clusterServers.setPassword(redisPassword);
}
//修改命令超时时间为40s，默认3s
clusterServers.setTimeout(40000);
//修改连接超时时间为50s，默认10s
clusterServers.setConnectTimeout(50000);
clusterServers.setCheckSlotsCoverage(false);
} else {
// 单机模式
SingleServerConfig singleServer = config.useSingleServer();
singleServer.setAddress("redis://" + redisHost + ":" + redisPort);
if (StringUtils.hasText(redisPassword)) {
singleServer.setPassword(redisPassword);
}
singleServer.setTimeout(40000);
singleServer.setConnectTimeout(50000);
}
}
RedissonClient redissonClient = null;
try {
redissonClient = Redisson.create(config);
} catch (Exception e) {
Log.error("初始化Redis失败", e);
}
return redissonClient;
}
@ConditionalOnProperty(name = "ruijie.tool.redis-enable", havingValue = "true")
@Bean
public CacheManager cacheManager(RedissonClient redissonClient) {
Map config = new HashMap();
//开辟命名空间，过期设置为1小时，连接最大存活时间为30分钟
config.put("springCache", new CacheConfig(60 * 60 * 1000, 30 * 60 * 1000));
return new RedissonSpringCacheManager(redissonClient, config);
}
框架层改造完了之后，现在只有一个入口即Redisson，只要Redisson稳住，就不会出现问题。那么该如何稳住呢，为了解决第二个问题，也就是Redis集群挂了还能正常启动，我的第一个想法是通过Spring的Bean生命周期入手，尝试开始。按照Redisson官方推荐的加载方式使用Redisson.create(config)创建一个RedissonClient接口的实现类，那么我们在代码中引入这个接口的话，Spring就会自动帮我们填充这个自定义的实现类。
组件可插拔
问题来了，如果这个Redis挂了的话，能不能启动呢？当然是肯定不能启动的，因为我现在就遇到这个问题了，但是为了模拟这个情况，我随便输入几个地址。
这里很明显的提示了如果要创建cacheManager(SpringCache核心类)，必须要提供一个RedissonClient的实现类。顺带一提，Redisson开启懒加载在这是无意义的，因为CacheManager cacheManager(RedissonClient redissonClient)这里必须要注入一个才能初始化，并不是想象中项目启动，业务代码中使用缓存调接口才初始化。
最开始我在做组件的时候，如果我想排掉三方包中的依赖，我的首选是实现BeanDefinitionRegistryPostProcessor接口，在Spring的Bean生命周期初始化后根据名字主动移除该Bean。这个不能解决我的问题，因为我这是在Bean初始化的时候就失败了。
常见的框架排Bean的方法，还有一招暴力的，直接在依赖配置文件中干掉这个框架依赖或者在SpringBoot启动类上排掉这个依赖的自动配置类。我可以选择将Redis整个依赖单独拆分出来成为一个新的组件，这样也能算是一个可插拔的设计，并且能让臃肿的框架包轻量一些。但是我不喜欢，所以我还是采用了配置类的方式，刚好Spring有足够的定制化配置，我选择了@ConditionalOnProperty(name = "ruijie.tool.redis-enable", havingValue = "true")这个注解。意思是配置文件中，只有配置ruijie.tool.redis-enable=true时才初始化被注解的Bean。
用法很简单，在框架中所有用到Redis的Bean上统统加上这个注解即可。但是同样存在一个限制，那就是项目代码中不能允许引入Redis相关类，引入的话同样会导致启动失败，因为框架中所有Redis相关的Bean都没有初始化。
尝试
因为还是没能从根源上解决无法启动的问题，于是我有了极端的想法，能不能注册一个假的Redis？首先我尝试在@Bean注解的Redisson注册方法中，提供一个null，妄图通过一个空的Bean来保证至少启动时不报错。经测试无效，当传入null时，Spring根本不加载该Bean，和上面初始化失败时报一样的错误，即找不到一个实现类。
因为是用Redisson初始化，所以我想着能不能通过构造方法做一个不进行初始化连接的Bean注入到Spring，跟一下Redisson的初始化源码
发现走的还是构造方法，点到构造方法后发现是protected修饰，还没有开放的重载
org.redisson.config.ConfigSupport#createConnectionManager这个方法就是创建连接的方法，连不上就报错......这条路算是堵死了
在多次尝试后，放弃了，毕竟从本质上来讲，Redis是个NoSql，就是个数据库。极端一点，类比MySQL，我本来就拿你当数据库使用，你项目数据库都挂了，那启动起来还有什么业务意义呢？与其思考这个，不如保证高可用，当然本期讨论的不是这个。
另一个问题
接下来要解决另一个问题，如何保证项目运行中Redis断连还能正常使用，同时保证老代码的改动最小。
之前也说过，解决缓存挂掉的最好方法就是捕获异常后转数据库连接，但是从框架层来讲，我哪里知道你连的是什么数据库，更不知道你要的是什么数据了。因此得转换思路，记得之前看过一个技术理论，如果遇到不好解决的问题，那么最便捷的方法就是加一个中间层。所以我选择了二级缓存，第二级缓存是本地缓存，用的caffeine，本质上是个ConcurrentHashMap，这肯定挂不了了。新建一个统一的缓存类，以后项目中引入该缓存类即可。
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import java.time.Duration;
/**
* spring redis 工具类
*
* @author ruoyi
**/
@Component
@Slf4j
@ConditionalOnProperty(name = "ruijie.tool.redis-enable",havingValue = "true")
public class CachePlusTool {
@Autowired
public RedissonClient redissonClient;
/**
* 本地缓存设置时不需要太长TTL以及太大容量，会占用过多内存造成OOM，得不偿失
* 设置初始容量为1000，最大容量为10000，普通业务足够，对于过多元素的可采用json压缩为一个元素
* 设置默认超时时间为4小时的考虑是经验值，过长的缓存没有意义，用户几乎在白天操作，且夜晚很多项目需要跑定时，会占用大量内存
*/
private final Cache caffeine = Caffeine.newBuilder()
.initialCapacity(1000).expireAfterAccess(Duration.ofHours(4))
.maximumSize(10000).build();
/**
* 缓存基本的对象，Integer、String、实体类等，默认超时时间4小时
*
* @param key 缓存的键值
* @param value 缓存的值
*/
public void put(String key, T value) {
try {
redissonClient.getBucket(key).set(value, Duration.ofHours(4));
} catch (Exception e) {
log.error("Redis的Put连接失败，降级处理使用本地缓存Caffeine", e);
caffeine.put(key, value);
}
}
/**
* 重载版本，仅支持对redis的ttl设置，对caffeine无效
*
* @param key 缓存的键值
* @param value 缓存的值
* @param timeout 过期时间
*/
public void put(String key, T value, Duration timeout) {
try {
redissonClient.getBucket(key).set(value, timeout);
} catch (Exception e) {
log.error("Redis的Put连接失败，降级处理使用本地缓存Caffeine", e);
caffeine.put(key, value);
}
}
/**
* 获得缓存的基本对象。
* 可能返回null值--因为对于缓存来说类似于数据库，分不清是程序错误导致的null还是数据为空的null，所以需要开发人员自行判断
*
* @param key 缓存键值
* @return 缓存键值对应的数据
*/
public T get(String key) {
Object value = null;
try {
value = redissonClient.getBucket(key).get();
} catch (Exception e) {
log.error("Redis的Get连接失败，降级处理使用本地缓存Caffeine", e);
//没有返回null
value = caffeine.getIfPresent(key);
}
return (T) value;
}
}
上图为测试结果，在Redis尝试无望后转而使用本地缓存，并会打印日志进行提示。并且该方法对于老项目比较友好，基本上只需替换引入的Redis类为新的CachePlusTool即可，使用方式如下。
后续可以继续扩容该工具了，如果有特殊的需求，那么说明强依赖于Redis，那就得承担强依赖带来的问题，不在本方案考虑范围内。
Redis单机快速部署
测试主要是测了一个Redis在项目运行中，突然断掉的情况，这里按照官方流程快速安装一个单机Redis，命令不用改
官网下载安装包
wget download.redis.io/redis-stabl…
解压缩
tar -xzvf redis-stable.tar.gz
到解压后根目录
cd redis-stable
编译
make
安装
make install
修改redis.conf
bind XXXX(IP地址)
protected-mode no(关掉保护模式)
requirepass 123(编一个123的简单密码)
这个部署倒是出人意料的简单，照着官网我就随便弄弄，没想到就好了。
方案总结
代码规范
解决的问题是项目运行中Redis挂了，项目依旧能提供本地缓存或者转数据库连接的服务，接口不会挂掉
对于常见的缓存存取，使用框架提供的CachePlusTool即可，代码会发到DevOps上，有补充的自行填充
@Autowired
public CachePlusTool cachePlusTool;
cachePlusTool.put(key, value);
cachePlusTool.put(key, value, Duration.ofHours(4));
cachePlusTool.get(key);
如果有特殊需求，比如分布式锁，或者项目是多节点部署且对数据正确性要求高的接口，一定要捕获异常，有必要的转数据库连接，代码片如下
@Autowired
private RedissonClient redissonClient;
public String getProductLine(String itemNo) {
String cacheKey = "order:getProductLine:" + itemNo;
String cacheValue = null;
RBucket bucket = redissonClient.getBucket(cacheKey);
try {
cacheValue = bucket.get();
} catch (Exception e) {
//捕获异常记得打印日志
log.error("redis连接异常", e);
}
if (cacheValue != null) {
return cacheValue;
} else {
//有必要的话转数据库查询
String res = ptmErpMapper.getProductLine(itemNo);
bucket.set(res, 16, TimeUnit.HOURS);
return res;
}
}
解耦Redis
解决的问题是完全不依赖Redis，但是依赖了框架，想要解耦Redis，避免因为Redis挂了而导致服务无法启动。
导入依赖
com.ruijie
tool-spring-boot-starter
2.3.3
在配置文件中加上开关配置
ruijie.tool.redis-enable=false
如上配置就让框架中所有Redis相关的Bean不会初始化，也就是彻底解耦，不会受到Redis集群状态的影响。
微醺码头
本期微醺码头主要是针对以上方案的一个回顾与发散，讨论下方案本身的问题以及我的一些构想。
首先我想说的是二级缓存方案中的本地缓存，对于多节点部署的项目来说是存在问题的，打个比方，如果是A节点使用了本地缓存获得了数据a，但是数据库此时如果变成了b，那么B节点通过本地缓存就得到了b，这样多节点给前端的数据就会变成一会儿是a，一会儿是b。当然真实场景，为了保证缓存一致性，可以选择旁路缓存模式，及时更新，甚至使用延时双删策略保证较强的一致性。
关于Redis这层解耦，其实我还有个想法，既然都加了一个中间层，为啥不把这个中间层扩大呢？构建一个Redis独立微服务，统一管理所有的Redis，提供通用的熔断和降级处理。当然，这光是想想就觉得问题很多，更期待读者能给我带来更优质的想法和建议。
写在最后
本来没想写的，花了上周五和本周一的时间来解决这个问题，找了下网上没啥好的想法，索性就我来写个。因为要给团队内写个使用文档，考虑到部分同事希望我多讲讲原理，喜欢听，那我就多写写，反正也不碍事。有个同事还要让我搞技术分享会，那算了，给我整活，之前搞过几次，效果不好，还老是下班搞，这不行。写写博客吧，想学的人自然会学，不想学的按着头也不会学，顺其自然。我最近也是受到了我目标的激励，重新唤起了动力，加油，Fighting！！！未来模模糊糊，总是有点犹豫不决，那就先往前走，行动起来总没错，诸君共勉！
