缓存框架JetCache源码解析-缓存变更通知机制

为什么需要缓存变更通知机制？如果我们使用的是本地缓存或者多级缓存（本地缓存+远程缓存），当其中一个节点的本地缓存变更之后，为了保证缓存尽量的一致性，此时其他节点的本地缓存也需要去变更，这时候常用的做法就是需要接入一个通知机制，只要有节点的本地缓存变更了，那么这时候就广播出一个缓存变更消息，其他节点会去订阅这个消息，最终消费这个消息的时候把本地缓存的数据进行更新或删除。当然，通知机制的实现有很多，比如可以利用redis的发布订阅机制，也可以接入mq这种消息中间件作为消息通知，不过相比之下，mq就稍微重了点，所以JetCache则是使用redis的发布订阅机制去实现缓存变更通知机制

缓存监视器CacheMonitor

@FunctionalInterface
public interface CacheMonitor {

    void afterOperation(CacheEvent event);

}

如果我们想要监控缓存的每一次操作，或者在缓存的每一次操作的时候进行一些埋点扩展，那么JetCache提供了CacheMonitor接口可以让我们用户自己去进行一些自定义的实现，当然在JetCache内部也有默认的一些缓存监视器的实现：

DefaultCacheMonitor

该monitor主要是监控缓存的每一次操作，比如get，put，remove等，基于监控这些操作之下就可以采集到各种操作指标，例如缓存命中次数，获取缓存次数，加载数据次数等等，详细的采集的指标如下：

protected String cacheName;
protected long statStartTime;
protected long statEndTime;

protected long getCount;
protected long getHitCount;
protected long getMissCount;
protected long getFailCount;
protected long getExpireCount;
protected long getTimeSum;
protected long minGetTime = Long.MAX_VALUE;
protected long maxGetTime = 0;

protected long putCount;
protected long putSuccessCount;
protected long putFailCount;
protected long putTimeSum;
protected long minPutTime = Long.MAX_VALUE;
protected long maxPutTime = 0;

protected long removeCount;
protected long removeSuccessCount;
protected long removeFailCount;
protected long removeTimeSum;
protected long minRemoveTime = Long.MAX_VALUE;
protected long maxRemoveTime = 0;

protected long loadCount;
protected long loadSuccessCount;
protected long loadFailCount;
protected long loadTimeSum;
protected long minLoadTime = Long.MAX_VALUE;
protected long maxLoadTime = 0;

CacheNotifyMonitor

该monitor看名字大概就知道它的作用是用来监控缓存变更之后发送通知的，监控的缓存操作有4种，分别是put，putAll，remove，removeAll，这4个操作都是会让缓存产生变更的操作，代码如下：

@Override
public void afterOperation(CacheEvent event) {
    if (this.broadcastManager == null) {
        return;
    }
    AbstractCache absCache = CacheUtil.getAbstractCache(event.getCache());
    if (absCache.isClosed()) {
        return;
    }
    AbstractEmbeddedCache localCache = getLocalCache(absCache);
    if (localCache == null) {
        return;
    }
    
    // put事件
    if (event instanceof CachePutEvent) {
        CacheMessage m = new CacheMessage();
        m.setArea(area);
        m.setCacheName(cacheName);
        m.setSourceId(sourceId);
        CachePutEvent e = (CachePutEvent) event;
        m.setType(CacheMessage.TYPE_PUT);
        m.setKeys(new Object[]{convertKey(e.getKey(), localCache)});
        broadcastManager.publish(m);
    }
        // remove事件
    else if (event instanceof CacheRemoveEvent) {
        CacheMessage m = new CacheMessage();
        m.setArea(area);
        m.setCacheName(cacheName);
        m.setSourceId(sourceId);
        CacheRemoveEvent e = (CacheRemoveEvent) event;
        m.setType(CacheMessage.TYPE_REMOVE);
        m.setKeys(new Object[]{convertKey(e.getKey(), localCache)});
        broadcastManager.publish(m);
    }
        // putAll事件
    else if (event instanceof CachePutAllEvent) {
        CacheMessage m = new CacheMessage();
        m.setArea(area);
        m.setCacheName(cacheName);
        m.setSourceId(sourceId);
        CachePutAllEvent e = (CachePutAllEvent) event;
        m.setType(CacheMessage.TYPE_PUT_ALL);
        if (e.getMap() != null) {
            m.setKeys(e.getMap().keySet().stream().map(k -> convertKey(k, localCache)).toArray());
        }
        broadcastManager.publish(m);
    }
        // removeAll事件
    else if (event instanceof CacheRemoveAllEvent) {
        CacheMessage m = new CacheMessage();
        m.setArea(area);
        m.setCacheName(cacheName);
        m.setSourceId(sourceId);
        CacheRemoveAllEvent e = (CacheRemoveAllEvent) event;
        m.setType(CacheMessage.TYPE_REMOVE_ALL);
        if (e.getKeys() != null) {
            m.setKeys(e.getKeys().stream().map(k -> convertKey(k, localCache)).toArray());
        }
        broadcastManager.publish(m);
    }
}

缓存变更消息发布器/订阅器BroadcastManager

上面我们知道了JetCache中会通过CacheNotifyMonitor来监控缓存的变更，但是发送缓存变更的消息通知则是交给BroadcastManager去做的。BroadcastManager是一个抽象类，提供了两个抽象方法：

/**
 * 发布缓存变更消息
 * @param cacheMessage  cacheMessage
 */
public abstract CacheResult publish(CacheMessage cacheMessage);

/**
 * 订阅缓存变更消息
 */
public abstract void startSubscribe();

publish方法是发布广播缓存变更消息的方法，传入的CacheMessage参数就是缓存变更消息，而startSubscribe方法则是订阅缓存变更消息的，这两个方法具体由子类进行实现，目前JetCache中主要提供了4种实现，这4种实现都是基于redis的订阅发布机制实现的，只是使用的redis客户端不一样的区别

LettuceBroadcastManager
RedisBroadcastManager
RedissonBroadcastManager
SpringDataBroadcastManager

我们这里可以以RedissonBroadcastManager为例，去看它是怎样实现缓存变更消息通知机制的

public class RedissonBroadcastManager extends BroadcastManager {
    private static final Logger logger = LoggerFactory.getLogger(RedissonBroadcastManager.class);
    private final RedissonCacheConfig<?, ?> config;
    private final String channel;
    private final RedissonClient client;
    private volatile int subscribeId;

    private final ReentrantLock reentrantLock = new ReentrantLock();

    public RedissonBroadcastManager(final CacheManager cacheManager, final RedissonCacheConfig<?, ?> config) {
        super(cacheManager);
        checkConfig(config);
        this.config = config;
        this.channel = config.getBroadcastChannel();
        this.client = config.getRedissonClient();
    }

    @Override
    public void startSubscribe() {
        reentrantLock.lock();
        try {
            if (this.subscribeId == 0 && Objects.nonNull(this.channel) && !this.channel.isEmpty()) {
                this.subscribeId = this.client.getTopic(this.channel)
                .addListener(byte[].class, (channel, msg) -> processNotification(msg, this.config.getValueDecoder()));
            }
        }finally {
            reentrantLock.unlock();
        }
    }


    @Override
    public void close() {
        reentrantLock.lock();
        try {
            final int id;
            if ((id = this.subscribeId) > 0 && Objects.nonNull(this.channel)) {
                this.subscribeId = 0;
                try {
                    this.client.getTopic(this.channel).removeListener(id);
                } catch (Throwable e) {
                    logger.warn("unsubscribe {} fail", this.channel, e);
                }
            }
        }finally {
            reentrantLock.unlock();
        }
    }

    @Override
    public CacheResult publish(final CacheMessage cacheMessage) {
        try {
            if (Objects.nonNull(this.channel) && Objects.nonNull(cacheMessage)) {
                final byte[] msg = this.config.getValueEncoder().apply(cacheMessage);
                this.client.getTopic(this.channel).publish(msg);
                return CacheResult.SUCCESS_WITHOUT_MSG;
            }
            return CacheResult.FAIL_WITHOUT_MSG;
        } catch (Throwable e) {
            SquashedLogger.getLogger(logger).error("jetcache publish error", e);
            return new CacheResult(e);
        }
    }
}

首先当调用Cache实例的put，putAll，remove，removeAll这些方法的时候，就会去触CacheNotifyMonitor，CacheNotifyMonitor这时候就会去创建一个缓存变更信息然后交给RedissonBroadcastManager去调用publish方法，在publish方法就会使用redisson的发布api去广播出这条缓存变更消息，当然订阅也是使用redisson的订阅api，当监听到有缓存变更消息的时候，这时候就会去回调processNotification方法，该方法是父类BroadcastManager的方法：

protected void processNotification(byte[] message, Function<byte[], Object> decoder) {
    try {
        if (message == null) {
            logger.error("notify message is null");
            return;
        }
        Object value = decoder.apply(message);
        if (value == null) {
            logger.error("notify message is null");
            return;
        }
        if (value instanceof CacheMessage) {
            processCacheMessage((CacheMessage) value);
        } else {
            logger.error("the message is not instance of CacheMessage, class={}", value.getClass());
        }
    } catch (Throwable e) {
        SquashedLogger.getLogger(logger).error("receive cache notify error", e);
    }
}

private void processCacheMessage(CacheMessage cacheMessage) {
    // 条件成立：说明这个缓存消息是自己发送的，这时候不用处理
    if (sourceId.equals(cacheMessage.getSourceId())) {
        return;
    }
    
    // 根据area和cacheName从缓存管理器中获取到对应的Cache实例
    Cache cache = cacheManager.getCache(cacheMessage.getArea(), cacheMessage.getCacheName());
    if (cache == null) {
        logger.warn("Cache instance not exists: {},{}", cacheMessage.getArea(), cacheMessage.getCacheName());
        return;
    }
    
    // 如果这个Cache实例不是一个多级缓存的Cache实例，那么就直接return，因为这里主要是针对多级缓存的情况下进行处理的
    Cache absCache = CacheUtil.getAbstractCache(cache);
    if (!(absCache instanceof MultiLevelCache)) {
        logger.warn("Cache instance is not MultiLevelCache: {},{}", cacheMessage.getArea(), cacheMessage.getCacheName());
        return;
    }
    
    // 获取到多级缓存中所有的Cache实例
    Cache[] caches = ((MultiLevelCache) absCache).caches();
    // 获取到缓存有变动的key
    Set<Object> keys = Stream.of(cacheMessage.getKeys()).collect(Collectors.toSet());
    // 遍历所有的Cache实例
    for (Cache c : caches) {
        Cache localCache = CacheUtil.getAbstractCache(c);
        // 只针对本地缓存的Cache实例，把缓存有变动的key从本地缓存中移除
        if (localCache instanceof AbstractEmbeddedCache) {
            ((AbstractEmbeddedCache) localCache).__removeAll(keys);
        } else {
            break;
        }
    }
}

最终会调用到processCacheMessage方法，需要注意的是如果是自己发送的消息则不需要再去消费，所以会使用一个sourceId去进行过滤，并且还会去判断当前的Cache实例是否是一个多级缓存实例，如果是多级缓存，那么就把这个多级缓存种的所有本地缓存都取出来，然后最后根据缓存变更消息里面的key把对应的本地缓存都remove掉，可能有人问为什么这里对本地缓存进行remove而不是update呢？因为如果使用update的话，还需要考虑并发的场景，比如这个key进行了两次更新，先后会发布两次缓存变更消息，但是订阅者去消费这两个消息的时候并不能去保证它们的先后顺序，此时就会有可能造成更新顺序的问题，但是使用remove就可能完全避免这种并发问题了

总结

在JetCache中提供了对于缓存操作的后置扩展接口，也叫做缓存监控器，我们可以自己去实现自己的缓存监控器来对缓存操作之后做一些自定义的功能，默认地JetCache提供了两个缓存监视器的实现，一个是采集缓存操作的一些指标信息DefaultCacheMonitor，一个是在缓存变更之后发送缓存变更通知的CacheNotifyMonitor，其中CacheNotifyMonitor在监控到缓存变更之后会去使用BroadcastManager去进行消息的发布订阅，BroadcastManager是发布者也是订阅者，在订阅消费的时会根据缓存变更消息里面的keys去把当前Cache实例的本地缓存进行remove，也就是依赖于这样的一个缓存变更消息通知机制，就可以保证当使用多级缓存的时候，多节点间的本地缓存尽量达成一致

秒客网

缓存框架JetCache源码解析-缓存变更通知机制

缓存监视器CacheMonitor

缓存变更消息发布器/订阅器BroadcastManager

总结

相关文章