关于java:Redis-60-客户端缓存

4次阅读

共计 4642 个字符，预计需要花费 12 分钟才能阅读完成。

不难发现，咱们常常将 Redis 作为零碎的缓存服务，但你有没有发现。在咱们每次操作 Redis 时，都须要发送网络申请。这样就防止不了网络的开销。但如何解决这个问题呢？咱们引入了本地缓存来解决此问题。查问逻辑从先前的间接查问转变为：先通过查问本地缓存，不存在再去近程查找而后设置到本地缓存 – 实用于分布式客户端缓存。

有没有感觉像咱们应用过的本地缓存 Guava、Caffeine 等一样？有啥特地的？这里 Redis 6 引入了 Tracking 机制。它的特别之处在于当某个节点的 key 对应的值发现变动或者生效后，会从服务端发送音讯告诉给其它节点。使得其它节点同步批改本地缓存。此时你可能又会想到，这不就是 Redis 音讯的公布订阅吗？我早就应用过了。有点相似，但也有所不同，Redis 把所有都帮咱们做好了，咱们这里只须要拿来间接用即可。方才提到的 Tracking 有三种模式，别离为：一般模式、播送模式、转发模式。留神：因为须要反对 Redis 服务端音讯推送，Redis 实现了 RESP3 协定。为了兼容以前的 RESP2 协定，故而引入了 转发模式，它外部通过Redis 音讯的公布与订阅实现。这里不做细讲，感兴趣的小伙伴能够查阅相干材料。

开启 tracking 模式，默认敞开。
开启 tracking 模式后，Redis 服务端会记录每个客户端申请过的 key，当 key 对应的值发生变化时，会发送生效音讯给客户端。

与一般模式的区别在于 Redis 服务端无需记录客户端申请过的 key, 而是当 key 对应的值发生变化时，发送音讯到客户端。
客户端能够只监听特定前缀 key 的音讯。

那么，咱们在业务中该如何抉择适合的模式呢？

这里大家能够依据本人的须要，一般模式会在服务端存储客户端申请过的 key 信息，会占用服务端的内存。而播送模式则会给所有客户端发送音讯，当然咱们也能够只关注特定前缀的音讯。要求咱们标准的命名 key。

在理解了两种模式之后，接下来通过 demo 加深下印象。

在 maven 我的项目中引入 Redis 依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

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 与下面的配置相似，间接上代码，就不做过多解析。

/**
 * 构建 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();}
    });
}

喜爱本博主的文章欢送关注我的公众号【我的极简博客】，浏览更多文章。

正文完