Merge pull request qiujiayu#73 from zhuangzhuangdashen/feature/lizhuangzhuang_addFixRateRefreshSupport

Feature/lizhuangzhuang add fix rate refresh support

Author: qiujiayu

pom.xml

@@ -164,7 +164,6 @@
       <version>2.1</version>
       <optional>true</optional>
     </dependency>
-
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>

src/main/java/com/jarvis/cache/AutoLoadHandler.java

@@ -7,12 +7,13 @@
 import com.jarvis.cache.to.CacheKeyTO;
 import com.jarvis.cache.to.CacheWrapper;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 
 import java.lang.reflect.Method;
 import java.util.Arrays;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadLocalRandom;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.*;
 
 /**
  * 用于处理自动加载缓存，sortThread 从autoLoadMap中取出数据，然后通知threads进行处理。
@@ -60,6 +61,13 @@ public class AutoLoadHandler {
      */
     private final AutoLoadConfig config;
 
+
+    private static ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;
+
+    public static ScheduledThreadPoolExecutor getScheduledThreadPoolExecutor() {
+        return scheduledThreadPoolExecutor;
+    }
+
     /**
      * @param cacheHandler 缓存的set,get方法实现类
      * @param config       配置
@@ -75,6 +83,8 @@ public AutoLoadHandler(CacheHandler cacheHandler, AutoLoadConfig config) {
             this.sortThread = new Thread(new SortRunnable());
             this.sortThread.setDaemon(true);
             this.sortThread.start();
+            // init fixed rate refresh process thread pool
+            initScheduledThreadPoolExecutor();
             for (int i = 0; i < this.config.getThreadCnt(); i++) {
                 this.threads[i] = new Thread(new AutoLoadRunnable());
                 this.threads[i].setName(THREAD_NAME_PREFIX + i);
@@ -89,6 +99,10 @@ public AutoLoadHandler(CacheHandler cacheHandler, AutoLoadConfig config) {
         }
     }
 
+    private void initScheduledThreadPoolExecutor() {
+        scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(10);
+    }
+
     public int getSize() {
         if (null != autoLoadMap) {
             return autoLoadMap.size();
@@ -146,6 +160,29 @@ public AutoLoadTO putIfAbsent(CacheKeyTO cacheKey, CacheAopProxyChain joinPoint,
         if (null == autoLoadMap) {
             return null;
         }
+
+        // 如果fixRateUpdateCache注解字段不为空,则走固定刷新逻辑
+        if(StringUtils.isNotEmpty(cache.fixRateUpdateCache())) {
+            AutoLoadTO autoLoadTO = autoLoadMap.get(cacheKey);
+            if (null != autoLoadTO) {
+                autoLoadMap.remove(cacheKey);
+            }
+
+            DeepClone deepClone = new DeepClone(joinPoint, cache).invoke();
+            if (deepClone.is()) return null;
+
+            Object[] arguments = deepClone.getArguments();
+            autoLoadTO = new AutoLoadTO(cacheKey, joinPoint, arguments, cache, Integer.MAX_VALUE);
+            // 设置过期时间为永不过期
+            autoLoadTO.setExpire(Integer.MAX_VALUE);
+
+            boolean openFixRateUpdateCache = fixRateUpdateCacheIfNeeded(joinPoint.getMethod().getName(), autoLoadTO);
+            if (openFixRateUpdateCache) {
+                return null;
+            }
+        }
+
+        // 走老逻辑-》交由AutoLoad处理
         AutoLoadTO autoLoadTO = autoLoadMap.get(cacheKey);
         if (null != autoLoadTO) {
             return autoLoadTO;
@@ -185,6 +222,120 @@ public AutoLoadTO putIfAbsent(CacheKeyTO cacheKey, CacheAopProxyChain joinPoint,
         return null;
     }
 
+    private boolean fixRateUpdateCacheIfNeeded(String methodName, AutoLoadTO autoLoadTO) {
+        if (null == autoLoadTO || autoLoadTO.getCache() == null ||
+                StringUtils.isEmpty(autoLoadTO.getCache().fixRateUpdateCache())) {
+            return false;
+        }
+
+        // 如果配置固定频率刷新 则缓存有效期为永久
+        doExecute(methodName, autoLoadTO);
+        return true;
+    }
+
+    private void doExecute(String methodName, AutoLoadTO autoLoadTO) {
+        // 解析fixRateUpdateCache Timer表达式
+        String updateCacheCronExpression = autoLoadTO.getCache().fixRateUpdateCache();
+        if (!updateCacheCronExpression.contains(",")) {
+            log.error("不符合规则的频率表达式{}", updateCacheCronExpression);
+            return;
+        }
+        long delay;
+        long period;
+        try {
+            String[] split = updateCacheCronExpression.split(",");
+            delay = Long.parseLong(split[0]);
+            period = Long.parseLong(split[1]);
+        } catch (Exception e) {
+            log.error("not matched cron expression-{}", updateCacheCronExpression);
+            return;
+        }
+        scheduledThreadPoolExecutor.scheduleWithFixedDelay(new FixRateUpdateCacheTask(autoLoadTO), delay,
+                period, TimeUnit.SECONDS);
+        log.info("register fix rate refresh task——method-{}, rate-{}", methodName, updateCacheCronExpression);
+    }
+
+    class FixRateUpdateCacheTask implements Runnable{
+        private AutoLoadTO autoLoadTO;
+
+        public FixRateUpdateCacheTask(AutoLoadTO autoLoadTO) {
+            this.autoLoadTO = autoLoadTO;
+        }
+
+        @Override
+        public void run() {
+            // 执行更新 依然复用"拿来主义"
+            Cache cache = autoLoadTO.getCache();
+            log.debug("执行定时刷新缓存任务, {}", cache.fixRateUpdateCache());
+            CacheWrapper<Object> result = null;
+            if (config.isCheckFromCacheBeforeLoad()) {
+                try {
+                    Method method = autoLoadTO.getJoinPoint().getMethod();
+                    result = cacheHandler.get(autoLoadTO.getCacheKey(), method);
+                } catch (Exception ex) {
+                    log.error(ex.getMessage(), ex);
+                }
+
+                if (null != result) {
+                    autoLoadTO.setExpire(result.getExpire());
+                    if (result.getLastLoadTime() > autoLoadTO.getLastLoadTime()) {
+                        autoLoadTO.setLastLoadTime(result.getLastLoadTime());
+                        return;
+                    }
+                }
+            }
+            CacheAopProxyChain pjp = autoLoadTO.getJoinPoint();
+            CacheKeyTO cacheKey = autoLoadTO.getCacheKey();
+            DataLoader dataLoader;
+            if (config.isDataLoaderPooled()) {
+                DataLoaderFactory factory = DataLoaderFactory.getInstance();
+                dataLoader = factory.getDataLoader();
+            } else {
+                dataLoader = new DataLoader();
+            }
+            CacheWrapper<Object> newCacheWrapper = null;
+            long loadDataUseTime = 0L;
+            try {
+                newCacheWrapper = dataLoader.init(pjp, autoLoadTO, cacheKey, cache, cacheHandler).loadData()
+                        .getCacheWrapper();
+                loadDataUseTime = dataLoader.getLoadDataUseTime();
+            } catch (Throwable e) {
+                log.error(e.getMessage(), e);
+            } finally {
+                if (config.isDataLoaderPooled()) {
+                    DataLoaderFactory factory = DataLoaderFactory.getInstance();
+                    factory.returnObject(dataLoader);
+                }
+            }
+            // 如果数据加载失败，则把旧数据进行续租
+            if (null == newCacheWrapper && null != result) {
+                newCacheWrapper = new CacheWrapper<Object>(result.getCacheObject(), Integer.MAX_VALUE);
+            }
+            writeCacheAndSetLoadTime(cache, pjp, cacheKey, newCacheWrapper, loadDataUseTime, autoLoadTO);
+        }
+    }
+
+    /**
+     * 写入缓存并且设置上一次加载时间
+     * @param cache
+     * @param pjp
+     * @param cacheKey
+     * @param newCacheWrapper
+     * @param loadDataUseTime
+     * @param autoLoadTO
+     */
+    private void writeCacheAndSetLoadTime(Cache cache, CacheAopProxyChain pjp, CacheKeyTO cacheKey, CacheWrapper<Object> newCacheWrapper, long loadDataUseTime, AutoLoadTO autoLoadTO) {
+        try {
+            if (null != newCacheWrapper) {
+                cacheHandler.writeCache(pjp, autoLoadTO.getArgs(), cache, cacheKey, newCacheWrapper);
+                autoLoadTO.setLastLoadTime(newCacheWrapper.getLastLoadTime())
+                        .setExpire(newCacheWrapper.getExpire()).addUseTotalTime(loadDataUseTime);
+            }
+        } catch (Exception e) {
+            log.error(e.getMessage(), e);
+        }
+    }
+
     /**
      * 获取自动加载队列，如果是web应用，建议把自动加载队列中的数据都输出到页面中，并增加一些管理功能。
      *
@@ -252,7 +403,6 @@ public void run() {
     }
 
     class AutoLoadRunnable implements Runnable {
-
         @Override
         public void run() {
             while (running) {
@@ -372,17 +522,46 @@ private void loadCache(AutoLoadTO autoLoadTO) {
                     int newExpire = AUTO_LOAD_MIN_EXPIRE + 60;
                     newCacheWrapper = new CacheWrapper<Object>(result.getCacheObject(), newExpire);
                 }
+                writeCacheAndSetLoadTime(cache, pjp, cacheKey, newCacheWrapper, loadDataUseTime, autoLoadTO);
+            }
+        }
+    }
+
+    private class DeepClone {
+        private boolean myResult;
+        private CacheAopProxyChain joinPoint;
+        private Cache cache;
+        private Object[] arguments;
+
+        public DeepClone(CacheAopProxyChain joinPoint, Cache cache) {
+            this.joinPoint = joinPoint;
+            this.cache = cache;
+        }
+
+        boolean is() {
+            return myResult;
+        }
+
+        public Object[] getArguments() {
+            return arguments;
+        }
+
+        public DeepClone invoke() {
+            if (cache.argumentsDeepcloneEnable()) {
                 try {
-                    if (null != newCacheWrapper) {
-                        cacheHandler.writeCache(pjp, autoLoadTO.getArgs(), cache, cacheKey, newCacheWrapper);
-                        autoLoadTO.setLastLoadTime(newCacheWrapper.getLastLoadTime())
-                                .setExpire(newCacheWrapper.getExpire()).addUseTotalTime(loadDataUseTime);
-                    }
+                    // 进行深度复制
+                    arguments = (Object[]) cacheHandler.getCloner().deepCloneMethodArgs(joinPoint.getMethod(),
+                            joinPoint.getArgs());
                 } catch (Exception e) {
                     log.error(e.getMessage(), e);
+                    myResult = true;
+                    return this;
                 }
+            } else {
+                arguments = joinPoint.getArgs();
             }
+            myResult = false;
+            return this;
         }
     }
-
 }

src/main/java/com/jarvis/cache/DataLoader.java

@@ -11,6 +11,8 @@
 import com.jarvis.cache.to.ProcessingTO;
 import lombok.extern.slf4j.Slf4j;
 
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+
 /**
  * 数据加载器
  *
@@ -139,36 +141,51 @@ public DataLoader loadData() throws Throwable {
     private void doFirstRequest(ProcessingTO processingTO) throws Throwable {
         ILock distributedLock = cacheHandler.getLock();
         // 开启分布式锁
-        if (null != distributedLock && cache.lockExpire() > 0) {
-            String lockKey = cacheKey.getLockKey();
-            long startWait = processingTO.getStartTime();
-            do {
-                // 获得分布式锁
-                if (distributedLock.tryLock(lockKey, cache.lockExpire())) {
-                    try {
-                        getData();
-                    } finally {
-                        distributedLock.unlock(lockKey);
+        Throwable throwable = null;
+        try {
+            if (null != distributedLock && cache.lockExpire() > 0) {
+                String lockKey = cacheKey.getLockKey();
+                long startWait = processingTO.getStartTime();
+                do {
+                    // 获得分布式锁
+                    if (distributedLock.tryLock(lockKey, cache.lockExpire())) {
+                        try {
+                            getData();
+                        } finally {
+                            distributedLock.unlock(lockKey);
+                        }
+                        break;
+                    }
+                    int tryCnt = 20;
+                    // 没有获得锁时，定时缓存尝试获取数据
+                    for (int i = 0; i < tryCnt; i++) {
+                        cacheWrapper = cacheHandler.get(cacheKey, pjp.getMethod());
+                        if (null != cacheWrapper) {
+                            break;
+                        }
+                        Thread.sleep(10);
                     }
-                    break;
-                }
-                int tryCnt = 20;
-                // 没有获得锁时，定时缓存尝试获取数据
-                for (int i = 0; i < tryCnt; i++) {
-                    cacheWrapper = cacheHandler.get(cacheKey, pjp.getMethod());
                     if (null != cacheWrapper) {
                         break;
                     }
-                    Thread.sleep(10);
-                }
-                if (null != cacheWrapper) {
-                    break;
+                } while (System.currentTimeMillis() - startWait < cache.waitTimeOut());
+                if (null == cacheWrapper) {
+                    throw new LoadDataTimeOutException("load data for key \"" + cacheKey.getCacheKey() + "\" timeout(" + cache.waitTimeOut() + " ms).");
                 }
-            } while (System.currentTimeMillis() - startWait < cache.waitTimeOut());
-            if (null == cacheWrapper) {
-                throw new LoadDataTimeOutException("load data for key \"" + cacheKey.getCacheKey() + "\" timeout(" + cache.waitTimeOut() + " ms).");
             }
-        } else {
+        } catch (Throwable e) {
+            if (cache.openLockDown()) {
+                throwable = e;
+                // 关闭分布式锁
+                cacheHandler.setLock(null);
+                log.error("分布式锁异常，强制停止使用分布式锁!", throwable);
+            } else {
+                // 否则抛异常
+                log.error("分布式锁异常!", e);
+                throw e;
+            }
+        }
+        if (throwable != null) {
             getData();
         }
         // 本地缓存

src/main/java/com/jarvis/cache/annotation/Cache.java

@@ -1,6 +1,7 @@
 package com.jarvis.cache.annotation;
 
 import com.jarvis.cache.type.CacheOpType;
+import lombok.Setter;
 
 import java.lang.annotation.Documented;
 import java.lang.annotation.ElementType;
@@ -71,6 +72,16 @@
      */
     boolean autoload() default false;
 
+    /**
+     * 以固定频率的方式刷新缓存 (补充expire无限大时无法依靠alarmTime频繁刷新缓存的不足)
+     * 格式 “initDelay,period” 默认单位s 【初始延迟时长,执行周期】
+     *              如【5，10】意为5s后开始以10s为周期执行刷新任务
+     *      为了向后兼容, 当alarmTime存在时优先解析alarmTime
+     *      暂不支持自定义+扩展
+     * @return 固定表达式 “initDelay,period”
+     */
+    String fixRateUpdateCache() default "";
+
     /**
      * 自动缓存的条件，可以为空，返回 true 或者 false，如果设置了此值，autoload() 就失效，例如：null !=
      * #args[0].keyword，当第一个参数的keyword属性为null时设置为自动加载。
@@ -122,6 +133,13 @@
      */
     int lockExpire() default 10;
 
+    /**
+     * 是否开启锁降级
+     *      默认不开启;
+     *      如果开启，当分布式锁抛异常时不使用分布式锁
+     */
+    boolean openLockDown() default false;
+
     /**
      * 是否打开对参数进行深度复制,默认是true,是为了避免外部改变参数值。如果确保不被修改，最好是设置为false,这样性能会更高。
      *