不难发现，我们经常将 Redis 作为系统的缓存服务，但你有没有发现。在我们每次操作 Redis 时，都需要发送网络请求。这样就避免不了网络的开销。但如何解决这个问题呢？我们引入了本地缓存来解决此问题。查询逻辑从先前的直接查询转变为：先通过查询本地缓存，不存在再去远程查找然后设置到本地缓存 - 适用于分布式客户端缓存。
有没有感觉像我们使用过的本地缓存 Guava、Caffeine 等一样？有啥特别的 ？这里 Redis 6 引入了 Tracking 机制。它的特别之处在于当某个节点的 key 对应的值发现变化或者失效后，会从服务端发送消息通知给其它节点。使得其它节点同步修改本地缓存。此时你可能又会想到，这不就是 Redis 消息的发布订阅吗？我早就使用过了。有点类似，但也有所不同，Redis 把一切都帮我们做好了，我们这里只需要拿来直接用即可。刚才提到的Tracking有三种模式，分别为：普通模式、广播模式、转发模式。注意：由于需要支持 Redis 服务端消息推送，Redis 实现了 RESP3 协议。为了兼容以前的 RESP2 协议，故而引入了转发模式，它内部通过Redis 消息的发布与订阅实现。这里不做细讲，感兴趣的小伙伴可以查阅相关资料。
1、普通模式
- 开启
tracking模式，默认关闭。 - 开启
tracking模式后，Redis服务端会记录每个客户端请求过的key，当 key 对应的值发生变化时，会发送失效消息给客户端。
2、广播模式
- 与普通模式的区别在于
Redis服务端无需记录客户端请求过的key, 而是当 key 对应的值发生变化时，发送消息到客户端。 - 客户端可以只监听特定前缀 key 的消息。
那么，我们在业务中该如何选择合适的模式呢？
这里大家可以根据自己的需要，普通模式会在 服务端 存储客户端请求过的 key 信息，会占用服务端的内存。而广播模式则会给所有客户端发送消息，当然我们也可以只关注特定前缀的消息。要求我们规范的命名 key 。
在了解了两种模式之后，接下来通过 demo 加深下印象。
在 maven 项目中引入 Redis 依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
Lettuce 实现方式
- 1、创建
RedisClient
RedisURI redisURI= RedisURI.builder()
.withHost("127.0.0.1")
.withPort(6379)
.withPassword("123456".toCharArray()) // 没有密码此处需注释
.build();
RedisClient client = RedisClient.create(redisURI);
StatefulRedisConnection<String, String> connect = client.connect();
- 2 构建
CacheFrontend客户端缓存
Map<String, String> map = new ConcurrentHashMap<>();
CacheAccessor<String, String> mapCacheAccessor = CacheAccessor.forMap(map);
CacheFrontend<String, String> cacheFrontend = ClientSideCaching.enable(mapCacheAccessor,connect,TrackingArgs.Builder.enabled().bcast().prefixes("user","order").noloop());
解析：通过 map 作为我们本地缓存存储。connect 为上一步得到的 redis client 连接。TrackingArgs.Builder.enabled() 启用 Redis Tracking 机制。bcast() 方法为广播模式，prefixes() 则是对特定 key 前缀的过滤。noloop() 不接收连接自身修改 key 消息。去掉 bcast() 等相关广播模式配置则为普通模式。接下来就可以直接使用 cacheFrontend 进行缓存操作了。
Spring Boot 实现方式
由于Spring Boot 与上面的配置类似，直接上代码，就不做过多解析。
/**
* 构建 CacheFrontend
*
* @param redisConnectionFactory
* @return
*/
@Bean
public <K, V> CacheFrontend<K, V> cacheFrontend(RedisConnectionFactory redisConnectionFactory) {
StatefulRedisConnection redisConnect = this.getRedisConnect(redisConnectionFactory);
if (redisConnect == null) {
return null;
}
CacheAccessor<K, V> mapCacheAccessor = CacheAccessor.forMap(new ConcurrentHashMap<>());
CacheFrontend cacheFrontend = ClientSideCaching.enable(
mapCacheAccessor,
redisConnect,
// 使用广播的方式，无需在服务端储存客户端的 key 信息
TrackingArgs.Builder.enabled().bcast().noloop()
);
return cacheFrontend;
}
/**
* 构建 redis client
*
* @param redisConnectionFactory
* @return
*/
private StatefulRedisConnection getRedisConnect(RedisConnectionFactory redisConnectionFactory) {
if (redisConnectionFactory instanceof LettuceConnectionFactory) {
AbstractRedisClient nativeClient = ((LettuceConnectionFactory) redisConnectionFactory).getNativeClient();
if (nativeClient instanceof RedisClient) {
return ((RedisClient) nativeClient).connect();
}
}
return null;
}
- 拓展
了解过它源码的小伙伴可能会发现，Redis 服务端推送消息到客户端是把客户端的本地缓存直接删除了，然后下次查询的时候直接作用到 Redis 查询最新值，然后再设置到 本地缓存。这样一来本地缓存就存在没有实时更新的情况。有实现实时更新效果的方式吗？答案是有的，通过反射删除所有默认的监听，然后添加一个自定义监听，当监听到 key 失效的消息时，更新本地 Map 。
private static final String INVALIDATE = "invalidate";
private StatefulRedisConnection<K, V> redisConnect;
private CacheAccessor<K, V> mapCacheAccessor;
private RedisCommands<K, V> redisCommands;
/**
* usage: RedisPushListenerConfig.Builder.autoConfiguration(redisConnect, mapCacheAccessor);
*/
public static class Builder {
public static <K, V> void autoConfiguration(StatefulRedisConnection<K, V> redisConnect, CacheAccessor<K, V> mapCacheAccessor) {
new RedisPushListenerConfig(redisConnect, mapCacheAccessor);
}
}
private RedisPushListenerConfig(StatefulRedisConnection<K, V> redisConnect, CacheAccessor<K, V> mapCacheAccessor) {
this.redisConnect = redisConnect;
this.mapCacheAccessor = mapCacheAccessor;
this.redisCommands = redisConnect.sync();
this.removeDefaultListeners();
this.addNewListener();
}
/**
* 通过反射删除所有默认监听
* {@link io.lettuce.core.support.caching.ClientSideCaching#create(CacheAccessor, RedisCache)} e}
*/
private void removeDefaultListeners() {
try {
Field field = StatefulRedisConnectionImpl.class.getDeclaredField("pushHandler");
field.setAccessible(true);
PushHandler pushHandler = (PushHandler) field.get(this.redisConnect);
List<PushListener> pushListeners = (CopyOnWriteArrayList) pushHandler.getPushListeners();
Iterator<PushListener> iterator = pushListeners.iterator();
while (iterator.hasNext()) {
pushListeners.remove(iterator.next());
}
} catch (NoSuchFieldException | IllegalAccessException e) {
log.error(e.getMessage(), e);
}
}
/**
* 添加一个自定义监听，键失效时 更新本地 Map
*/
private void addNewListener() {
this.redisConnect.addListener(message -> {
if (INVALIDATE.equals(message.getType())) {
List<Object> content = message.getContent(StringCodec.UTF8::decodeKey);
List<K> keys = (List<K>) content.get(1);
new Thread(() -> {
for (K key : keys) {
V newVal = this.redisCommands.get(key);
mapCacheAccessor.put(key, newVal);
}
}).start();
}
});
}
喜欢本博主的文章欢迎关注我的公众号【我的极简博客】，阅读更多文章。
