公司网站怎么维护,做遗嘱的网站有哪些,做贸易 公司网站放哪里,网站设计制做报价作为一个缓存框架#xff0c;JetCache支持多级缓存#xff0c;也就是本地缓存和远程缓存#xff0c;但是不管是使用着两者中的哪一个或者两者都进行使用#xff0c;缓存的实时性一直都是我们需要考虑的问题#xff0c;通常我们为了尽可能地保证缓存的实时性#xff0c;都… 作为一个缓存框架JetCache支持多级缓存也就是本地缓存和远程缓存但是不管是使用着两者中的哪一个或者两者都进行使用缓存的实时性一直都是我们需要考虑的问题通常我们为了尽可能地保证缓存的实时性都会去采用一些策略比如更新数据库时同时删除缓存又或者是使用缓存双删等策略而在JetCache中它还支持定时地去刷新缓存又进一步地能够保证缓存的实时性 功能使用
编程式 QuickConfig idQc QuickConfig.newBuilder(:user:cache:id:).cacheType(CacheType.BOTH).expire(Duration.ofHours(2)).refreshPolicy(RefreshPolicy.newPolicy(1,TimeUnit.HOURS)).syncLocal(true).build();
idUserCache cacheManager.getOrCreateCache(idQc);
在编程式的方式中通过refreshPolicy属性配置指定缓存刷新的时间间隔比如上面定义的是每隔1小时刷新一次缓存
声明式
Cached(name :user:cache:id:, cacheType CacheType.BOTH, key #userId, cacheNullValue true)
CacheRefresh(refresh 60, timeUnit TimeUnit.MINUTES)
public User findById(Long userId) {return userMapper.findById(userId);
}
声明式的方式JetCache中提供的是CacheRefresh注解在该注解中的refresh属性中可以配置缓存的刷新时间
源码解析
RefreshCache的继承体系 CacheCache实例的顶级接口定义了Cache实例常用的一些api例如getput等等ProxyCache继承于Cache接口通过名字可以知道ProxyCache就是用于代理某一个Cache实例的而它也提供了一个getTargetCache方法该方法可以返回这个被代理的Cache实例SimpleProxyCache它是ProxyCache接口的一个实现里面有一个Cache实例的成员变量这个Cache实例其实就是被代理的Cache实例而SimpleProxyCache中的方法也很简单都是实现Cache接口以及ProxyCache接口的方法在实现的方法中调用成员变量Cache实例的同名方法而已其实就是一个代理模式LoadingCache通过名字可以知道这个Cache具有加载数据的功能所谓的加载数据就是从某一处数据源比如数据库es等进行数据加载然后再放到整个被代理的Cache实例中RefreshCache缓存定时刷新的核心实现在里面实现了通过定时器去进行缓存的刷新
缓存定时刷新任务RefreshTask
/*** 缓存刷新任务以key为维度每一个key对应一个RefreshTask* 这个任务主要做的事以key为维度每隔一段时间配置的refreshMillis执行一次通过loader加载出数据更新到缓存中* 如果使用的是多级缓存那么就会先loader数据到远程缓存中然后再更新到本地缓存中*/
class RefreshTask implements Runnable {private Object taskId;/*** 目标刷新的key*/private K key;/*** loader*/private CacheLoaderK, V loader;/*** 上一次访问时间*/private long lastAccessTime;private ScheduledFuture future;RefreshTask(Object taskId, K key, CacheLoaderK, V loader) {this.taskId taskId;this.key key;this.loader loader;}private void cancel() {logger.debug(cancel refresh: {}, key);future.cancel(false);taskMap.remove(taskId);}/*** 通过loader加载出数据如果有必要则还需把加载出的数据放到Cache中*/private void load() throws Throwable {CacheLoaderK, V l loader null ? config.getLoader() : loader;if (l ! null) {// 获取到一个ProxyLoader这个ProxyLoader会去对loader进行代理在原来loader加载数据的功能上增加了其他功能l CacheUtil.createProxyLoader(cache, l, eventConsumer);// 加载数据V v l.load(key);// 条件成立说明加载出来的数据需要更新到Cache中if (needUpdate(v, l)) {// 假如这个被代理的cache是一个多级缓存那么此时本地缓存和远程缓存都会被更新到了cache.PUT(key, v);}}}/*** 该方法的作用是如果目标key已经到达了缓存刷新时间则会使用loader去进行加载这个key的数据加载出来的数据会放入到缓存中* 在使用loader加载数据的过程中会以key为维度去加一把分布式锁目的就是为了防止多个机器节点同时进行loader因为只需要一台机器实例去loader数据到远程缓存即可* 如果抢到分布式锁成功那么就会把加载到的数据放到Cache缓存实例中如果抢不到分布式锁则说明已经有其他机器loader数据到远程缓存中了这时候就不用loader数据了* 但是还需要注意的是如果这时候使用的是多级缓存则还需要更新本地缓存所以就会从远程缓存中获取到数据然后更新本地缓存** param concreteCache 远程缓存Cache实例* param currentTime 当前时间用于判断key是否到达刷新时间了*/private void externalLoad(final Cache concreteCache, final long currentTime) throws Throwable {byte[] newKey ((AbstractExternalCache) concreteCache).buildKey(key);byte[] lockKey combine(newKey, LOCK_KEY_SUFFIX);// 获取到刷新缓存时加锁的过期时间long loadTimeOut RefreshCache.this.config.getRefreshPolicy().getRefreshLockTimeoutMillis();// 刷新缓存的时间间隔long refreshMillis config.getRefreshPolicy().getRefreshMillis();byte[] timestampKey combine(newKey, TIMESTAMP_KEY_SUFFIX);// 从缓存中获取到这个key上一次的刷新时间CacheGetResult refreshTimeResult concreteCache.GET(timestampKey);// 是否应该对这个key进行刷新boolean shouldLoad false;// 如果这个key存在并且这个key已经到达了下一次的刷新时间那么就需要对它进行刷新if (refreshTimeResult.isSuccess()) {shouldLoad currentTime Long.parseLong(refreshTimeResult.getValue().toString()) refreshMillis;}// 如果这个key并不存在那么也应该对它进行刷新else if (refreshTimeResult.getResultCode() CacheResultCode.NOT_EXISTS) {shouldLoad true;}// 条件成立不需要对这个key进行刷新if (!shouldLoad) {// 如果是多级缓存那么此时从最后一级缓存中获取数据刷新到前面层级的缓存中if (multiLevelCache) {refreshUpperCaches(key);}return;}// 抢到分布式锁之后会去执行这个runnable// runnable做的事情通过loader加载出数据如果有必要则还需把加载出的数据放到Cache实例中Runnable r () - {try {// 通过loader加载出数据如果有必要则还需把加载出的数据放到缓存中load();// 更新这个key最近的一次刷新时间concreteCache.put(timestampKey, String.valueOf(System.currentTimeMillis()));} catch (Throwable e) {throw new CacheException(refresh error, e);}};// 尝试去获取锁如果获取到锁就执行rboolean lockSuccess concreteCache.tryLockAndRun(lockKey, loadTimeOut, TimeUnit.MILLISECONDS, r);// 获取锁失败说明有其他线程获取到锁了也就说明其他线程正在加载数据到远程缓存中if (!lockSuccess multiLevelCache) {// 如果被代理的Cache实例是多级缓存的话还需要从远程缓存中获取到数据然后把数据刷新到上层的本地缓存// 延迟执行因为此时其他线程还在加载数据到远程缓存中JetCacheExecutor.heavyIOExecutor().schedule(() - refreshUpperCaches(key), (long) (0.2 * refreshMillis), TimeUnit.MILLISECONDS);}}/*** 从最后一级缓存中获取数据刷新到前面层级的缓存中** param key key*/private void refreshUpperCaches(K key) {MultiLevelCacheK, V targetCache (MultiLevelCacheK, V) getTargetCache();Cache[] caches targetCache.caches();int len caches.length;// 从最后一级缓存中获取到这个key的缓存数据CacheGetResult cacheGetResult caches[len - 1].GET(key);if (!cacheGetResult.isSuccess()) {return;}// 刷新前面层级的缓存for (int i 0; i len - 1; i) {caches[i].PUT(key, cacheGetResult.getValue());}}Overridepublic void run() {try {if (config.getRefreshPolicy() null || (loader null !hasLoader())) {cancel();return;}long now System.currentTimeMillis();// 获取到停止缓存刷新的时间间隔long stopRefreshAfterLastAccessMillis config.getRefreshPolicy().getStopRefreshAfterLastAccessMillis();if (stopRefreshAfterLastAccessMillis 0) {// 条件成立上一次访问这个key的时间 停止缓存刷新的时间间隔 当前时间// 也就是说这个key距离上一次访问已经有一段时间了这时候可能意味着这个key并不是一个频繁被访问的key所以此时可以把它的缓存刷新定时任务取消掉if (lastAccessTime stopRefreshAfterLastAccessMillis now) {logger.debug(cancel refresh: {}, key);// 取消这个key的缓存刷新定时任务cancel();return;}}logger.debug(refresh key: {}, key);// 获取到被代理的Cache实例如果被代理的Cache实例是多级缓存的Cache实例那么就返回最后一级缓存的Cache实例Cache concreteCache concreteCache();// 如果这个Cache实例是远程缓存实例那么就执行externalLoadif (concreteCache instanceof AbstractExternalCache) {// 在externalLoad方法中会使用loader加载数据到远程缓存中externalLoad(concreteCache, now);}// 如果这个Cache实例不是远程缓存实例那么就直接load数据到本地缓存中else {load();}} catch (Throwable e) {logger.error(refresh error: key key, e);}}
}
RefreshTask实现了Runnable接口也就是说它是运行在子线程中的而每一个RefreshTask任务都只会服务于一个key也就是说当我们访问了某一个key的数据那么这个key就会创建出对应的RefreshTask去对它进行定时的刷新
1run
public void run() {try {if (config.getRefreshPolicy() null || (loader null !hasLoader())) {cancel();return;}long now System.currentTimeMillis();// 获取到停止缓存刷新的时间间隔long stopRefreshAfterLastAccessMillis config.getRefreshPolicy().getStopRefreshAfterLastAccessMillis();if (stopRefreshAfterLastAccessMillis 0) {// 条件成立上一次访问这个key的时间 停止缓存刷新的时间间隔 当前时间// 也就是说这个key距离上一次访问已经有一段时间了这时候可能意味着这个key并不是一个频繁被访问的key所以此时可以把它的缓存刷新定时任务取消掉if (lastAccessTime stopRefreshAfterLastAccessMillis now) {logger.debug(cancel refresh: {}, key);// 取消这个key的缓存刷新定时任务cancel();return;}}logger.debug(refresh key: {}, key);// 获取到被代理的Cache实例如果被代理的Cache实例是多级缓存的Cache实例那么就返回最后一级缓存的Cache实例Cache concreteCache concreteCache();// 如果这个Cache实例是远程缓存实例那么就执行externalLoadif (concreteCache instanceof AbstractExternalCache) {// 在externalLoad方法中会使用loader加载数据到远程缓存中externalLoad(concreteCache, now);}// 如果这个Cache实例不是远程缓存实例那么就直接load数据到本地缓存中else {load();}} catch (Throwable e) {logger.error(refresh error: key key, e);}
}
首先会去获取stopRefreshAfterLastAccessMillis这个配置值这个配置值表示停止缓存刷新的时间间隔也就是说如果一个key的上一次访问时间距离当前时间已经超过了这个时间值了那么就可以停止掉这个key的缓存刷新定时任务了接着就会去通过loader加载器去加载数据了但是加载的时候需要去判断当前被代理的Cache是否有使用远程缓存如果没有使用远程缓存就会执行load方法
/*** 通过loader加载出数据如果有必要则还需把加载出的数据放到Cache中*/
private void load() throws Throwable {CacheLoaderK, V l loader null ? config.getLoader() : loader;if (l ! null) {// 获取到一个ProxyLoader这个ProxyLoader会去对loader进行代理在原来loader加载数据的功能上增加了其他功能l CacheUtil.createProxyLoader(cache, l, eventConsumer);// 加载数据V v l.load(key);// 条件成立说明加载出来的数据需要更新到Cache中if (needUpdate(v, l)) {// 假如这个被代理的cache是一个多级缓存那么此时本地缓存和远程缓存都会被更新到了cache.PUT(key, v);}}
}
protected boolean needUpdate(V loadedValue, CacheLoaderK, V loader) {if (loadedValue null !config.isCacheNullValue()) {return false;}// 是否禁止缓存的更新默认不禁止if (loader.vetoCacheUpdate()) {return false;}return true;
}
load方法很简单其实就是调用loader的load方法去加载数据然后再把加载数据的数据交给needUpdate方法在needUpdate方法中会判断是否设置缓存null值的配置如果加载出来的数据为null并且设置了允许缓存null值那么就会进一步再判断loader的vetoCacheUpdate方法该方法表示加载出来的时候是否允许更新到缓存中默认为false表示允许更新到缓存中最后再把加载的数据放到被代理的Cache实例中
2externalLoad
/*** 该方法的作用是如果目标key已经到达了缓存刷新时间则会使用loader去进行加载这个key的数据加载出来的数据会放入到缓存中* 在使用loader加载数据的过程中会以key为维度去加一把分布式锁目的就是为了防止多个机器节点同时进行loader因为只需要一台机器实例去loader数据到远程缓存即可* 如果抢到分布式锁成功那么就会把加载到的数据放到Cache缓存实例中如果抢不到分布式锁则说明已经有其他机器loader数据到远程缓存中了这时候就不用loader数据了* 但是还需要注意的是如果这时候使用的是多级缓存则还需要更新本地缓存所以就会从远程缓存中获取到数据然后更新本地缓存** param concreteCache 远程缓存Cache实例* param currentTime 当前时间用于判断key是否到达刷新时间了*/
private void externalLoad(final Cache concreteCache, final long currentTime) throws Throwable {byte[] newKey ((AbstractExternalCache) concreteCache).buildKey(key);byte[] lockKey combine(newKey, LOCK_KEY_SUFFIX);// 获取到刷新缓存时加锁的过期时间long loadTimeOut RefreshCache.this.config.getRefreshPolicy().getRefreshLockTimeoutMillis();// 刷新缓存的时间间隔long refreshMillis config.getRefreshPolicy().getRefreshMillis();byte[] timestampKey combine(newKey, TIMESTAMP_KEY_SUFFIX);// 从缓存中获取到这个key上一次的刷新时间CacheGetResult refreshTimeResult concreteCache.GET(timestampKey);// 是否应该对这个key进行刷新boolean shouldLoad false;// 如果这个key存在并且这个key已经到达了下一次的刷新时间那么就需要对它进行刷新if (refreshTimeResult.isSuccess()) {shouldLoad currentTime Long.parseLong(refreshTimeResult.getValue().toString()) refreshMillis;}// 如果这个key并不存在那么也应该对它进行刷新else if (refreshTimeResult.getResultCode() CacheResultCode.NOT_EXISTS) {shouldLoad true;}// 条件成立不需要对这个key进行刷新if (!shouldLoad) {// 如果是多级缓存那么此时从最后一级缓存中获取数据刷新到前面层级的缓存中if (multiLevelCache) {refreshUpperCaches(key);}return;}// 抢到分布式锁之后会去执行这个runnable// runnable做的事情通过loader加载出数据如果有必要则还需把加载出的数据放到Cache实例中Runnable r () - {try {// 通过loader加载出数据如果有必要则还需把加载出的数据放到缓存中load();// 更新这个key最近的一次刷新时间concreteCache.put(timestampKey, String.valueOf(System.currentTimeMillis()));} catch (Throwable e) {throw new CacheException(refresh error, e);}};// 尝试去获取锁如果获取到锁就执行rboolean lockSuccess concreteCache.tryLockAndRun(lockKey, loadTimeOut, TimeUnit.MILLISECONDS, r);// 获取锁失败说明有其他线程获取到锁了也就说明其他线程正在加载数据到远程缓存中if (!lockSuccess multiLevelCache) {// 如果被代理的Cache实例是多级缓存的话还需要从远程缓存中获取到数据然后把数据刷新到上层的本地缓存// 延迟执行因为此时其他线程还在加载数据到远程缓存中JetCacheExecutor.heavyIOExecutor().schedule(() - refreshUpperCaches(key), (long) (0.2 * refreshMillis), TimeUnit.MILLISECONDS);}
}
当使用了远程缓存时就会执行externalLoad方法去加载数据那么为什么使用了远程缓存之后就需要单独使用externalLoad方法去加载而不是load方法呢原因就是我们部署的服务如果是有多个节点的话对于每一个节点来说相同的key它们都会去起一个RefreshTask子线程去从数据库中加载数据然后put到远程缓存中但是这个操作需要每一个节点都执行吗显然不需要只需要其中一个节点把数据库的数据加载到远程缓存中然后其他节点在这个过程中阻塞等待缓存结果即可
首先会判断这个key是否到达了刷新时间了如果还没到达刷新时间那么就把远程缓存的数据刷新到上层缓存通常是本地缓存即可反之如果key到达了刷新时间了那么就会去尝试获取锁因为上面也说到了只需要一个节点执行就行了如果获取锁成功了那么就执行load方法从数据库中加载数据到被代理的Cache实例中如果这个被代理的Cache实例是一个多级缓存那么它的本地缓存和远程缓存都会被刷新了如果没有抢到锁说明已经有其他节点抢到锁去加载数据了如果这时候是多级缓存那么此时就会延迟五分之一的刷新时间再去从远程缓存刷新数据到本地缓存中那么为什么要延迟执行呢这是因为如果抢不到锁就立刻去从远程缓存获取数据的话这时候加载数据的节点可能还没有加载完成也就是说远程缓存这时候还是旧的数据所以JetCache这里就适当地做了一下延迟但是这样还会导致另一个问题那就是这个延迟不就会导致节点间本地缓存不一致了吗单单从这个缓存定时刷新机制来看是会导致这个问题的但是JetCache为了保证多节点间的本地缓存尽可能一致性还有一个缓存更新通知机制作为兜底有了这个缓存更新通知机制上述的问题也就能够解决了
RefreshTask任务的添加
上面讲了整个RefreshTask任务的执行流程那么它是在那里被添加的呢 private ConcurrentHashMapObject, RefreshTask taskMap new ConcurrentHashMap();
在RefreshCache中有一个mapkey可以理解为就是key值value存放的这个key对应的RefreshTask任务
Override
public V get(K key) throws CacheInvokeException {// 如果配置缓存刷新策略并且也有设置了对应的loaderif (config.getRefreshPolicy() ! null hasLoader()) {// 给这个key添加一个缓存刷新的定时任务addOrUpdateRefreshTask(key, null);}return super.get(key);
}Override
public MapK, V getAll(Set? extends K keys) throws CacheInvokeException {// 如果配置缓存刷新策略并且也有设置了对应的loaderif (config.getRefreshPolicy() ! null hasLoader()) {// 给每一个key都添加一个缓存刷新的定时任务for (K key : keys) {addOrUpdateRefreshTask(key, null);}}return super.getAll(keys);
}
而在RefreshCache中会去重写get和getAll这个两个方法当我们每次去访问一个key对应的缓存数据的时候如果这个key在taskMap中没有对应的RefreshTask那么就会调用addOrUpdateRefreshTask方法创建一个
/*** 给指定的key添加一个缓存刷新定时任务RefreshTask如果这个key已经存在对应的RefreshTask那么就更新一下这个RefreshTask的访问时间* param key key* param loader loader*/
protected void addOrUpdateRefreshTask(K key, CacheLoaderK, V loader) {// 获取配置的缓存刷新策略RefreshPolicy refreshPolicy config.getRefreshPolicy();// 如果没有配置缓存刷新策略那么直接returnif (refreshPolicy null) {return;}// 获取缓存刷新的时间间隔long refreshMillis refreshPolicy.getRefreshMillis();if (refreshMillis 0) {// 根据key获取到对应的taskIdObject taskId getTaskId(key);// 根据taskId从taskMap中获取一个缓存刷新任务RefreshTask如果taskMap中没有则创建一个RefreshTask refreshTask taskMap.computeIfAbsent(taskId, tid - {logger.debug(add refresh task. interval{}, key{}, refreshMillis, key);// 创建一个RefreshTaskRefreshTask task new RefreshTask(taskId, key, loader);task.lastAccessTime System.currentTimeMillis();ScheduledFuture? future JetCacheExecutor.heavyIOExecutor().scheduleWithFixedDelay(task, refreshMillis, refreshMillis, TimeUnit.MILLISECONDS);task.future future;return task;});// 更新这个key的访问时间为当前的最新时间refreshTask.lastAccessTime System.currentTimeMillis();}
}
如果这个key在taskMap中已经存在对应的RefreshTask了那么就取出这个RefreshTask更新一下它的lastAccessTime即可在执行这个RefreshTask任务的时候如果配置了stopRefreshAfterLastAccessMillis的话就会通过去比较lastAccessTime来决定是否还需要定时更新这个key如果不需要则会把这个RefreshTask任务取消掉然后再从taskMap移除
总结
当我们去访问每一个key的时候JetCache都会对这个key生成一个RefreshTask任务然后会把这个RefreshTask任务交给定时任务调度器。在执行RefreshTask任务的过程中主要就是分为两种情况一种是当前使用了远程缓存一种是没有使用远程缓存如果使用了远程缓存那么此时会先去获取锁只有获取锁成功了才会去从数据库中加载数据然后刷新到远程缓存中如果没有抢到锁但是又使用了多级缓存本地缓存远程缓存那么这时候就会延迟一段时间才会去把远程缓存的时候刷新到本地缓存延迟的原因是因为抢到锁的节点这时候可能还没有完成数据的加载到远程缓存中而正是因为这个延迟就会有可能导致多节点间本地缓存数据不一致的情况针对这种情况JetCache则是通过缓存更新通知机制去进行兜底
