Sugus

V1

2022/12/20阅读:9主题:绿意

Redis 6.0 客户端缓存

不难发现,我们经常将 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();
        }
    });
}

喜欢本博主的文章欢迎关注我的公众号【我的极简博客】,阅读更多文章。

分类:

后端

标签:

Java

作者介绍

Sugus
V1