Pre Merge pull request !49 from buzhidaolvtu/master

xxl-job-admin/src/main/java/com/xxl/job/admin/core/route/ExecutorRouter.java

@@ -1,5 +1,6 @@
 package com.xxl.job.admin.core.route;
 
+import com.xxl.job.admin.core.model.XxlJobGroup;
 import com.xxl.job.core.biz.model.ReturnT;
 import com.xxl.job.core.biz.model.TriggerParam;
 import org.slf4j.Logger;
@@ -17,8 +18,20 @@ public abstract class ExecutorRouter {
      * route address
      *
      * @param addressList
-     * @return  ReturnT.content=address
+     * @return ReturnT.content=address
      */
     public abstract ReturnT<String> route(TriggerParam triggerParam, List<String> addressList);
 
+
+    /**
+     * route address within group
+     * @param group
+     * @param triggerParam
+     * @param addressList
+     * @return
+     */
+    public ReturnT<String> route(XxlJobGroup group, TriggerParam triggerParam, List<String> addressList) {
+        return route(triggerParam, addressList);
+    }
+
 }

xxl-job-admin/src/main/java/com/xxl/job/admin/core/route/ExecutorRouterHelper.java

@@ -0,0 +1,40 @@
+package com.xxl.job.admin.core.route;
+
+import com.xxl.job.admin.core.route.strategy.ExecutorRouteLFU;
+import com.xxl.job.admin.core.route.strategy.ExecutorRouteLRU;
+
+public class ExecutorRouterHelper {
+
+    /**
+     * 以executor为单位，更新LRU和LFU
+     *
+     * @param appname
+     * @param address
+     */
+    public static void updateRouteStats(String appname, String address) {
+
+        updateLRUStats(appname, address);
+
+        updateLFUStats(appname, address);
+    }
+
+    /**
+     * 统计LFU
+     *
+     * @param appname
+     * @param address
+     */
+    private static void updateLFUStats(String appname, String address) {
+        ExecutorRouteLFU.AppAddressPool.getOrCreate(appname).updateAddressStats(address);
+    }
+
+    /**
+     * @param appname
+     * @param address
+     */
+    private static void updateLRUStats(String appname, String address) {
+        ExecutorRouteLRU.AppAddressPool.getOrCreate(appname).updateAddressStats(address);
+    }
+
+
+}

xxl-job-admin/src/main/java/com/xxl/job/admin/core/route/strategy/ExecutorRouteLFU.java

@@ -10,70 +10,102 @@ import java.util.concurrent.ConcurrentMap;
 
 /**
  * 单个JOB对应的每个执行器，使用频率最低的优先被选举
- *      a(*)、LFU(Least Frequently Used)：最不经常使用，频率/次数
+ * a(*)、LFU(Least Frequently Used)：最不经常使用，频率/次数
- *      b、LRU(Least Recently Used)：最近最久未使用，时间
+ * b、LRU(Least Recently Used)：最近最久未使用，时间
- *
+ * <p>
  * Created by xuxueli on 17/3/10.
  */
 public class ExecutorRouteLFU extends ExecutorRouter {
 
-    private static ConcurrentMap<Integer, HashMap<String, Integer>> jobLfuMap = new ConcurrentHashMap<Integer, HashMap<String, Integer>>();
+    public static class AppAddressPool {
-    private static long CACHE_VALID_TIME = 0;
+        private static ConcurrentMap<String, AppAddressPool> APP_LFU_MANAGER = new ConcurrentHashMap();
 
-    public String route(int jobId, List<String> addressList) {
+        /**
+         * 每个APP name 实例化一个地址池
+         *
+         * @param appname
+         * @return
+         */
+        public static AppAddressPool getOrCreate(String appname) {
+            return APP_LFU_MANAGER.computeIfAbsent(appname, key -> new AppAddressPool());
+        }
 
-        // cache clear
+        private List<String> freeAddressList = new ArrayList<>();
-        if (System.currentTimeMillis() > CACHE_VALID_TIME) {
+
-            jobLfuMap.clear();
+        //key=address
-            CACHE_VALID_TIME = System.currentTimeMillis() + 1000*60*60*24;
+        private HashMap<String, Integer> usedAddressManager = new HashMap<>();
+
+        private synchronized void clear(String appname) {
+            APP_LFU_MANAGER.remove(appname);
+            getOrCreate(appname);
         }
 
-        // lfu item init
+        public synchronized void updateAddressStats(String address) {
-        HashMap<String, Integer> lfuItemMap = jobLfuMap.get(jobId);     // Key排序可以用TreeMap+构造入参Compare；Value排序暂时只能通过ArrayList；
+            usedAddressManager.put(address, usedAddressManager.getOrDefault(address, 0) + 1);
-        if (lfuItemMap == null) {
-            lfuItemMap = new HashMap<String, Integer>();
-            jobLfuMap.putIfAbsent(jobId, lfuItemMap);   // 避免重复覆盖
         }
 
-        // put new
+        private synchronized String route(List<String> addressList) {
-        for (String address: addressList) {
+            HashMap<String, Integer> lfuItemMap = usedAddressManager;
-            if (!lfuItemMap.containsKey(address) || lfuItemMap.get(address) >1000000 ) {
+
-                lfuItemMap.put(address, new Random().nextInt(addressList.size()));  // 初始化时主动Random一次，缓解首次压力
+            // remove dead address
+            List<String> delKeys = new ArrayList<>();
+            for (String existKey : lfuItemMap.keySet()) {
+                if (!addressList.contains(existKey)) {
+                    delKeys.add(existKey);
+                }
             }
-        }
+            if (delKeys.size() > 0) {
-        // remove old
+                for (String delKey : delKeys) {
-        List<String> delKeys = new ArrayList<>();
+                    lfuItemMap.remove(delKey);
-        for (String existKey: lfuItemMap.keySet()) {
+                }
-            if (!addressList.contains(existKey)) {
-                delKeys.add(existKey);
             }
-        }
+
-        if (delKeys.size() > 0) {
+            // put new address into free list
-            for (String delKey: delKeys) {
+            freeAddressList.clear();
-                lfuItemMap.remove(delKey);
+            for (String address : addressList) {
+                if (!lfuItemMap.containsKey(address)) {
+                    freeAddressList.add(address);
+                }
             }
-        }
 
-        // load least userd count address
+            if (freeAddressList.size() > 0) {
-        List<Map.Entry<String, Integer>> lfuItemList = new ArrayList<Map.Entry<String, Integer>>(lfuItemMap.entrySet());
+                String freeAddress = freeAddressList.get(0);
-        Collections.sort(lfuItemList, new Comparator<Map.Entry<String, Integer>>() {
+                freeAddressList.remove(0);
-            @Override
+                return freeAddress;
-            public int compare(Map.Entry<String, Integer> o1, Map.Entry<String, Integer> o2) {
-                return o1.getValue().compareTo(o2.getValue());
             }
-        });
 
-        Map.Entry<String, Integer> addressItem = lfuItemList.get(0);
+            // load least userd count address
-        String minAddress = addressItem.getKey();
+            List<Map.Entry<String, Integer>> lfuItemList = new ArrayList<Map.Entry<String, Integer>>(lfuItemMap.entrySet());
-        addressItem.setValue(addressItem.getValue() + 1);
+            Collections.sort(lfuItemList, new Comparator<Map.Entry<String, Integer>>() {
+                @Override
+                public int compare(Map.Entry<String, Integer> o1, Map.Entry<String, Integer> o2) {
+                    return o1.getValue().compareTo(o2.getValue());
+                }
+            });
+
+            Map.Entry<String, Integer> addressItem = lfuItemList.get(0);
+
+            return addressItem.getKey();
+        }
 
-        return addressItem.getKey();
     }
 
+    public String route(String appname, List<String> addressList) {
+        AppAddressPool appAddressPool = AppAddressPool.getOrCreate(appname);
+        // cache clear
+        if (System.currentTimeMillis() > CACHE_VALID_TIME) {
+            appAddressPool.clear(appname);
+            CACHE_VALID_TIME = System.currentTimeMillis() + 1000 * 60 * 60 * 24;
+        }
+
+        return appAddressPool.route(addressList);
+    }
+
+    private static long CACHE_VALID_TIME = 0;
+
     @Override
     public ReturnT<String> route(TriggerParam triggerParam, List<String> addressList) {
-        String address = route(triggerParam.getJobId(), addressList);
+        throw new RuntimeException("remove");
-        return new ReturnT<String>(address);
     }
 
 }

xxl-job-admin/src/main/java/com/xxl/job/admin/core/route/strategy/ExecutorRouteLRU.java

@@ -1,5 +1,6 @@
 package com.xxl.job.admin.core.route.strategy;
 
+import com.xxl.job.admin.core.model.XxlJobGroup;
 import com.xxl.job.admin.core.route.ExecutorRouter;
 import com.xxl.job.core.biz.model.ReturnT;
 import com.xxl.job.core.biz.model.TriggerParam;
@@ -12,65 +13,106 @@ import java.util.concurrent.ConcurrentMap;
 
 /**
  * 单个JOB对应的每个执行器，最久为使用的优先被选举
- *      a、LFU(Least Frequently Used)：最不经常使用，频率/次数
+ * a、LFU(Least Frequently Used)：最不经常使用，频率/次数
- *      b(*)、LRU(Least Recently Used)：最近最久未使用，时间
+ * b(*)、LRU(Least Recently Used)：最近最久未使用，时间
- *
+ * <p>
  * Created by xuxueli on 17/3/10.
  */
 public class ExecutorRouteLRU extends ExecutorRouter {
 
-    private static ConcurrentMap<Integer, LinkedHashMap<String, String>> jobLRUMap = new ConcurrentHashMap<Integer, LinkedHashMap<String, String>>();
+    public static class AppAddressPool {
-    private static long CACHE_VALID_TIME = 0;
+        private static ConcurrentMap<String, AppAddressPool> APP_LRU_MANAGER = new ConcurrentHashMap();
 
-    public String route(int jobId, List<String> addressList) {
+        /**
+         * 每个APP name 实例化一个地址池
+         *
+         * @param appname
+         * @return
+         */
+        public static AppAddressPool getOrCreate(String appname) {
+            return APP_LRU_MANAGER.computeIfAbsent(appname, key -> new AppAddressPool());
+        }
 
-        // cache clear
+        private List<String> freeAddressList = new ArrayList<>();
-        if (System.currentTimeMillis() > CACHE_VALID_TIME) {
+
-            jobLRUMap.clear();
+        /**
-            CACHE_VALID_TIME = System.currentTimeMillis() + 1000*60*60*24;
+         * LinkedHashMap
+         * a、accessOrder：true=访问顺序排序（get/put时排序）；false=插入顺序排期；
+         * b、removeEldestEntry：新增元素时将会调用，返回true时会删除最老元素；可封装LinkedHashMap并重写该方法，比如定义最大容量，超出是返回true即可实现固定长度的LRU算法；
+         */
+        private LinkedHashMap<String, String> usedAddressManager = new LinkedHashMap<String, String>(16, 0.75f, true);
+
+        private synchronized void clear(String appname) {
+            APP_LRU_MANAGER.remove(appname);
+            getOrCreate(appname);
         }
 
-        // init lru
+        public synchronized void updateAddressStats(String address) {
-        LinkedHashMap<String, String> lruItem = jobLRUMap.get(jobId);
+            usedAddressManager.put(address, address);
-        if (lruItem == null) {
-            /**
-             * LinkedHashMap
-             *      a、accessOrder：true=访问顺序排序（get/put时排序）；false=插入顺序排期；
-             *      b、removeEldestEntry：新增元素时将会调用，返回true时会删除最老元素；可封装LinkedHashMap并重写该方法，比如定义最大容量，超出是返回true即可实现固定长度的LRU算法；
-             */
-            lruItem = new LinkedHashMap<String, String>(16, 0.75f, true);
-            jobLRUMap.putIfAbsent(jobId, lruItem);
         }
 
-        // put new
+        private synchronized String route(List<String> addressList) {
-        for (String address: addressList) {
+            LinkedHashMap<String, String> lruItem = usedAddressManager;
-            if (!lruItem.containsKey(address)) {
+
-                lruItem.put(address, address);
+            // remove dead address
+            List<String> delKeys = new ArrayList<>();
+            for (String existKey : lruItem.keySet()) {
+                if (!addressList.contains(existKey)) {
+                    delKeys.add(existKey);
+                }
             }
-        }
+            if (delKeys.size() > 0) {
-        // remove old
+                for (String delKey : delKeys) {
-        List<String> delKeys = new ArrayList<>();
+                    lruItem.remove(delKey);
-        for (String existKey: lruItem.keySet()) {
+                }
-            if (!addressList.contains(existKey)) {
-                delKeys.add(existKey);
             }
-        }
+
-        if (delKeys.size() > 0) {
+            // put new address into free list
-            for (String delKey: delKeys) {
+            freeAddressList.clear();
-                lruItem.remove(delKey);
+            for (String address : addressList) {
+                if (!lruItem.containsKey(address)) {
+                    freeAddressList.add(address);
+                }
             }
+
+            if (freeAddressList.size() > 0) {
+                String freeAddress = freeAddressList.get(0);
+                freeAddressList.remove(0);
+                return freeAddress;
+            }
+
+            // load
+            String eldestKey = lruItem.entrySet().iterator().next().getKey();
+            String eldestValue = lruItem.get(eldestKey);
+            return eldestValue;
+
         }
 
-        // load
-        String eldestKey = lruItem.entrySet().iterator().next().getKey();
-        String eldestValue = lruItem.get(eldestKey);
-        return eldestValue;
     }
 
+    public String route(String appname, List<String> addressList) {
+        AppAddressPool appAddressPool = AppAddressPool.getOrCreate(appname);
+        // cache clear
+        if (System.currentTimeMillis() > CACHE_VALID_TIME) {
+            appAddressPool.clear(appname);
+            CACHE_VALID_TIME = System.currentTimeMillis() + 1000 * 60 * 60 * 24;
+        }
+
+        return appAddressPool.route(addressList);
+    }
+
+    private static long CACHE_VALID_TIME = 0;
+
     @Override
-    public ReturnT<String> route(TriggerParam triggerParam, List<String> addressList) {
+    public ReturnT<String> route(XxlJobGroup group, TriggerParam triggerParam, List<String> addressList) {
-        String address = route(triggerParam.getJobId(), addressList);
+        String address = route(group.getAppname(), addressList);
         return new ReturnT<String>(address);
     }
 
+
+    @Override
+    public ReturnT<String> route(TriggerParam triggerParam, List<String> addressList) {
+        throw new RuntimeException("remove");
+    }
+
 }

xxl-job-admin/src/main/java/com/xxl/job/admin/core/trigger/XxlJobTrigger.java

@@ -5,6 +5,7 @@ import com.xxl.job.admin.core.model.XxlJobGroup;
 import com.xxl.job.admin.core.model.XxlJobInfo;
 import com.xxl.job.admin.core.model.XxlJobLog;
 import com.xxl.job.admin.core.route.ExecutorRouteStrategyEnum;
+import com.xxl.job.admin.core.route.ExecutorRouterHelper;
 import com.xxl.job.admin.core.scheduler.XxlJobScheduler;
 import com.xxl.job.admin.core.util.I18nUtil;
 import com.xxl.job.core.biz.ExecutorBiz;
@@ -149,9 +150,11 @@ public class XxlJobTrigger {
                     address = group.getRegistryList().get(0);
                 }
             } else {
-                routeAddressResult = executorRouteStrategyEnum.getRouter().route(triggerParam, group.getRegistryList());
+                routeAddressResult = executorRouteStrategyEnum.getRouter().route(group, triggerParam, group.getRegistryList());
                 if (routeAddressResult.getCode() == ReturnT.SUCCESS_CODE) {
                     address = routeAddressResult.getContent();
+                    //以单个executor为范围，更新address的使用统计
+                    ExecutorRouterHelper.updateRouteStats(group.getAppname(), address);
                 }
             }
         } else {

xxl-job-admin/src/test/java/com/xxl/job/admin/core/route/ExecutorRouterHelperTest.java

@@ -0,0 +1,72 @@
+package com.xxl.job.admin.core.route;
+
+import com.xxl.job.admin.core.model.XxlJobGroup;
+import com.xxl.job.core.biz.model.ReturnT;
+import com.xxl.job.core.biz.model.TriggerParam;
+import org.junit.jupiter.api.Test;
+
+import java.util.Date;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
+class ExecutorRouterHelperTest {
+
+    @Test
+    public void test_LFU() throws InterruptedException {
+        XxlJobGroup group = new XxlJobGroup();
+        group.setAppname("setAppName");
+        group.setTitle("setTitle");
+        group.setAddressType(0);
+        group.setAddressList("191,192,193,194,195,196");
+        group.setUpdateTime(new Date());
+
+        TriggerParam triggerParam = new TriggerParam();
+        triggerParam.setJobId(1);
+
+
+        ConcurrentMap<String, AtomicInteger> addressStatsMap = new ConcurrentHashMap<>();
+        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(100, 100, 0, TimeUnit.HOURS, new LinkedBlockingDeque<>());
+        for (int i = 0; i < 1000; i++) {
+            threadPoolExecutor.execute(() -> {
+                ReturnT<String> route = ExecutorRouteStrategyEnum.LEAST_FREQUENTLY_USED.getRouter().route(group, triggerParam, group.getRegistryList());
+                if (route.getCode() == ReturnT.SUCCESS_CODE) {
+                    String address = route.getContent();
+                    AtomicInteger atomicInteger = addressStatsMap.computeIfAbsent(address, key -> new AtomicInteger(0));
+                    atomicInteger.incrementAndGet();
+                    //以单个executor为范围，更新address的使用统计
+                    ExecutorRouterHelper.updateRouteStats(group.getAppname(), address);
+                }
+            });
+        }
+        threadPoolExecutor.shutdown();
+        threadPoolExecutor.awaitTermination(1, TimeUnit.HOURS);
+        System.out.println(addressStatsMap.toString());
+    }
+
+    @Test
+    public void test_LRU() {
+        XxlJobGroup group = new XxlJobGroup();
+        group.setAppname("setAppName");
+        group.setTitle("setTitle");
+        group.setAddressType(0);
+        group.setAddressList("191,192,193,194,195,196");
+        group.setUpdateTime(new Date());
+
+        TriggerParam triggerParam = new TriggerParam();
+        triggerParam.setJobId(1);
+
+
+        ConcurrentMap<String, AtomicInteger> addressStatsMap = new ConcurrentHashMap<>();
+        for (int i = 0; i < 100; i++) {
+            ReturnT<String> route = ExecutorRouteStrategyEnum.LEAST_RECENTLY_USED.getRouter().route(group, triggerParam, group.getRegistryList());
+            if (route.getCode() == ReturnT.SUCCESS_CODE) {
+                String address = route.getContent();
+                AtomicInteger atomicInteger = addressStatsMap.computeIfAbsent(address, key -> new AtomicInteger(0));
+                atomicInteger.incrementAndGet();
+                //以单个executor为范围，更新address的使用统计
+                ExecutorRouterHelper.updateRouteStats(group.getAppname(), address);
+            }
+        }
+        System.out.println(addressStatsMap.toString());
+    }
+}