perf[scheduler]: 优化代码结构，并删除无用代码

event/src/main/java/com/zfoo/event/schema/EventRegisterProcessor.java

@@ -80,8 +80,8 @@ public class EventRegisterProcessor implements BeanPostProcessor {
                 var enhanceReceiverDefinition = EnhanceUtils.createEventReceiver(receiverDefinition);
                 EventBus.registerEventReceiver(eventClazz, enhanceReceiverDefinition);
             }
-        } catch (Exception e) {
-            throw new RuntimeException(e);
+        } catch (Throwable t) {
+            throw new RuntimeException(t);
         }
 
         return bean;

net/src/main/java/com/zfoo/net/consumer/registry/ZookeeperRegistry.java

@@ -27,7 +27,7 @@ import com.zfoo.protocol.util.AssertionUtils;
 import com.zfoo.protocol.util.IOUtils;
 import com.zfoo.protocol.util.JsonUtils;
 import com.zfoo.protocol.util.StringUtils;
-import com.zfoo.scheduler.SchedulerContext;
+import com.zfoo.scheduler.manager.SchedulerBus;
 import com.zfoo.util.ThreadUtils;
 import com.zfoo.util.net.HostAndPort;
 import io.netty.util.concurrent.FastThreadLocalThread;
@@ -317,7 +317,7 @@ public class ZookeeperRegistry implements IRegistry {
                 initConsumerCache();
             } catch (Exception e) {
                 logger.error("zookeeper初始化失败，等待[{}]秒，重新初始化", RETRY_SECONDS, e);
-                SchedulerContext.getSchedulerManager().schedule(new Runnable() {
+                SchedulerBus.schedule(new Runnable() {
                     @Override
                     public void run() {
                         initZookeeper();
@@ -441,7 +441,7 @@ public class ZookeeperRegistry implements IRegistry {
         }
 
         if (recheckFlag) {
-            SchedulerContext.getSchedulerManager().schedule(new Runnable() {
+            SchedulerBus.schedule(new Runnable() {
                 @Override
                 public void run() {
                     checkConsumer();

net/src/main/java/com/zfoo/net/dispatcher/manager/PacketBus.java

@@ -1,3 +1,16 @@
+/*
+ * Copyright (C) 2020 The zfoo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and limitations under the License.
+ */
+
 package com.zfoo.net.dispatcher.manager;
 
 import com.zfoo.event.model.event.IEvent;
@@ -111,8 +124,8 @@ public abstract class PacketBus {
                 var receiverDefinition = new PacketReceiverDefinition(bean, method, packetClazz, attachmentClazz);
                 var enhanceReceiverDefinition = EnhanceUtils.createPacketReceiver(receiverDefinition);
                 packetReceiverList[protocolId] = enhanceReceiverDefinition;
-            } catch (Exception e) {
-                throw new RuntimeException(e);
+            } catch (Throwable t) {
+                throw new RuntimeException(t);
             }
         }
     }

net/src/main/java/com/zfoo/net/util/SimpleCache.java

@@ -19,7 +19,7 @@ import com.github.benmanes.caffeine.cache.LoadingCache;
 import com.zfoo.event.manager.EventBus;
 import com.zfoo.protocol.collection.CollectionUtils;
 import com.zfoo.protocol.model.Pair;
-import com.zfoo.scheduler.manager.SchedulerManager;
+import com.zfoo.scheduler.manager.SchedulerBus;
 import org.checkerframework.checker.nullness.qual.NonNull;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
@@ -92,7 +92,7 @@ public class SimpleCache<K, V> {
                 });
 
 
-        SchedulerManager.getInstance().scheduleAtFixedRate(new Runnable() {
+        SchedulerBus.scheduleAtFixedRate(new Runnable() {
             @Override
             public void run() {
                 // 不在任务调度线程中执行耗时任务，因为任务调度线程只有一个线程池

orm/src/main/java/com/zfoo/orm/model/persister/CronOrmPersister.java

@@ -18,7 +18,7 @@ import com.zfoo.orm.OrmContext;
 import com.zfoo.orm.model.cache.EntityCaches;
 import com.zfoo.orm.model.vo.EntityDef;
 import com.zfoo.protocol.exception.ExceptionUtils;
-import com.zfoo.scheduler.SchedulerContext;
+import com.zfoo.scheduler.manager.SchedulerBus;
 import com.zfoo.scheduler.util.TimeUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -74,7 +74,7 @@ public class CronOrmPersister extends AbstractOrmPersister {
         }
 
         if (!OrmContext.isStop()) {
-            SchedulerContext.getSchedulerManager().schedule(new Runnable() {
+            SchedulerBus.schedule(new Runnable() {
                 @Override
                 public void run() {
                     if (!OrmContext.isStop()) {

orm/src/main/java/com/zfoo/orm/model/persister/TimeOrmPersister.java

@@ -18,7 +18,7 @@ import com.zfoo.orm.OrmContext;
 import com.zfoo.orm.model.cache.EntityCaches;
 import com.zfoo.orm.model.vo.EntityDef;
 import com.zfoo.protocol.util.StringUtils;
-import com.zfoo.scheduler.SchedulerContext;
+import com.zfoo.scheduler.manager.SchedulerBus;
 
 import java.util.concurrent.TimeUnit;
 
@@ -43,7 +43,7 @@ public class TimeOrmPersister extends AbstractOrmPersister {
 
     @Override
     public void start() {
-        SchedulerContext.getSchedulerManager().scheduleAtFixedRate(new Runnable() {
+        SchedulerBus.scheduleAtFixedRate(new Runnable() {
             @Override
             public void run() {
                 if (!OrmContext.isStop()) {

scheduler/src/main/java/com/zfoo/scheduler/SchedulerContext.java

@@ -14,8 +14,7 @@
 package com.zfoo.scheduler;
 
 import com.zfoo.protocol.util.ReflectionUtils;
-import com.zfoo.scheduler.manager.ISchedulerManager;
-import com.zfoo.scheduler.manager.SchedulerManager;
+import com.zfoo.scheduler.manager.SchedulerBus;
 import com.zfoo.scheduler.schema.SchedulerRegisterProcessor;
 import com.zfoo.util.ThreadUtils;
 import org.slf4j.Logger;
@@ -53,11 +52,6 @@ public class SchedulerContext implements ApplicationListener<ApplicationContextE
         return instance.applicationContext;
     }
 
-
-    public static ISchedulerManager getSchedulerManager() {
-        return SchedulerManager.getInstance();
-    }
-
     public static boolean isStop() {
         return stop;
     }
@@ -70,15 +64,10 @@ public class SchedulerContext implements ApplicationListener<ApplicationContextE
 
         stop = true;
 
-        ISchedulerManager schedulerManager = getSchedulerManager();
-        if (schedulerManager == null) {
-            return;
-        }
-
         try {
-            Field field = SchedulerManager.class.getDeclaredField("executor");
+            Field field = SchedulerBus.class.getDeclaredField("executor");
             ReflectionUtils.makeAccessible(field);
-            var executor = (ScheduledExecutorService) ReflectionUtils.getField(field, schedulerManager);
+            var executor = (ScheduledExecutorService) ReflectionUtils.getField(field, null);
             ThreadUtils.shutdown(executor);
         } catch (Throwable e) {
             logger.error("Scheduler thread pool failed shutdown.", e);

scheduler/src/main/java/com/zfoo/scheduler/manager/ISchedulerManager.java

@@ -1,40 +0,0 @@
-/*
- * Copyright (C) 2020 The zfoo Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
- * in compliance with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed
- * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and limitations under the License.
- */
-
-package com.zfoo.scheduler.manager;
-
-import java.util.concurrent.TimeUnit;
-
-/**
- * @author jaysunxiao
- * @version 3.0
- */
-public interface ISchedulerManager {
-
-    /**
-     * 不断执行的周期循环任务
-     */
-    void scheduleAtFixedRate(Runnable runnable, long period, TimeUnit unit);
-
-    /**
-     * 固定延迟执行的任务
-     */
-    void schedule(Runnable runnable, long delay, TimeUnit unit);
-
-
-    /**
-     * cron表达式执行的任务
-     */
-    void scheduleCron(Runnable runnable, String cron);
-
-}

scheduler/src/main/java/com/zfoo/scheduler/manager/SchedulerBus.java

@@ -15,16 +15,12 @@ package com.zfoo.scheduler.manager;
 
 import com.zfoo.protocol.collection.CollectionUtils;
 import com.zfoo.protocol.util.JsonUtils;
-import com.zfoo.protocol.util.ReflectionUtils;
-import com.zfoo.protocol.util.StringUtils;
 import com.zfoo.scheduler.SchedulerContext;
-import com.zfoo.scheduler.model.anno.Scheduler;
 import com.zfoo.scheduler.model.vo.SchedulerDefinition;
 import com.zfoo.scheduler.util.TimeUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.lang.reflect.Modifier;
 import java.util.Comparator;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -36,11 +32,9 @@ import java.util.concurrent.TimeUnit;
  * @author jaysunxiao
  * @version 3.0
  */
-public class SchedulerManager implements ISchedulerManager {
+public abstract class SchedulerBus {
 
-    private static final Logger logger = LoggerFactory.getLogger(SchedulerManager.class);
-
-    private static final SchedulerManager INSTANCE = new SchedulerManager();
+    private static final Logger logger = LoggerFactory.getLogger(SchedulerBus.class);
 
     private static final List<SchedulerDefinition> schedulerDefList = new CopyOnWriteArrayList<>();
 
@@ -56,10 +50,10 @@ public class SchedulerManager implements ISchedulerManager {
     private static long minSchedulerTriggerTimestamp = 0;
 
 
-    private static final long TRIGGER_MILLIS_INTERVAL = TimeUtils.MILLIS_PER_SECOND;
+    public static final long TRIGGER_MILLIS_INTERVAL = TimeUtils.MILLIS_PER_SECOND;
 
     /**
-     * scheduler默认只有一个单线程线程池
+     * scheduler默认只有一个单线程的线程池
      */
     private static final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(new SchedulerThreadFactory(1));
 
@@ -71,15 +65,9 @@ public class SchedulerManager implements ISchedulerManager {
             } catch (Exception e) {
                 logger.error("scheduler triggers an error.", e);
             }
-        }, TimeUtils.MILLIS_PER_SECOND, TRIGGER_MILLIS_INTERVAL, TimeUnit.MILLISECONDS);
+        }, 3 * TimeUtils.MILLIS_PER_SECOND, TRIGGER_MILLIS_INTERVAL, TimeUnit.MILLISECONDS);
     }
 
-    public static SchedulerManager getInstance() {
-        return INSTANCE;
-    }
-
-    private SchedulerManager() {
-    }
 
     private static long minSchedulerTriggerTimestamp() {
         var minSchedulerOptional = schedulerDefList.stream().min(Comparator.comparingLong(schedulerDef -> schedulerDef.getTriggerTimestamp()));
@@ -133,44 +121,16 @@ public class SchedulerManager implements ISchedulerManager {
         minSchedulerTriggerTimestamp = minSchedulerTriggerTimestamp();
     }
 
-    public void registerScheduler(Object bean) {
-        try {
-            var methods = ReflectionUtils.getMethodsByAnnoInPOJOClass(bean.getClass(), Scheduler.class);
-            for (var method : methods) {
-                var scheduler = method.getAnnotation(Scheduler.class);
-
-                var paramClazzs = method.getParameterTypes();
-                if (paramClazzs.length >= 1) {
-                    throw new IllegalArgumentException(StringUtils.format("[class:{}] [method:{}] can not have any parameters", bean.getClass(), method.getName()));
-                }
-
-                var methodName = method.getName();
-
-                if (!Modifier.isPublic(method.getModifiers())) {
-                    throw new IllegalArgumentException(StringUtils.format("[class:{}] [method:{}] must use 'public' as modifier!", bean.getClass().getName(), methodName));
-                }
-
-                if (Modifier.isStatic(method.getModifiers())) {
-                    throw new IllegalArgumentException(StringUtils.format("[class:{}] [method:{}] can not use 'static' as modifier!", bean.getClass().getName(), methodName));
-                }
-
-                if (!methodName.startsWith("cron")) {
-                    throw new IllegalArgumentException(StringUtils.format("[class:{}] [method:{}] must start with 'cron' as method name!"
-                            , bean.getClass().getName(), methodName));
-                }
-
-                var schedulerDef = SchedulerDefinition.valueOf(scheduler.cron(), bean, method);
-                schedulerDefList.add(schedulerDef);
-                minSchedulerTriggerTimestamp = minSchedulerTriggerTimestamp();
-            }
-        } catch (Throwable throwable) {
-            throw new RuntimeException(throwable);
-        }
+    public static void registerScheduler(SchedulerDefinition scheduler) {
+        schedulerDefList.add(scheduler);
+        minSchedulerTriggerTimestamp = minSchedulerTriggerTimestamp();
     }
 
 
-    @Override
-    public void scheduleAtFixedRate(Runnable runnable, long period, TimeUnit unit) {
+    /**
+     * 不断执行的周期循环任务
+     */
+    public static void scheduleAtFixedRate(Runnable runnable, long period, TimeUnit unit) {
         if (SchedulerContext.isStop()) {
             return;
         }
@@ -189,8 +149,11 @@ public class SchedulerManager implements ISchedulerManager {
         }, 0, period, unit);
     }
 
-    @Override
-    public void schedule(Runnable runnable, long delay, TimeUnit unit) {
+
+    /**
+     * 固定延迟执行的任务
+     */
+    public static void schedule(Runnable runnable, long delay, TimeUnit unit) {
         if (SchedulerContext.isStop()) {
             return;
         }
@@ -209,7 +172,9 @@ public class SchedulerManager implements ISchedulerManager {
         }, delay, unit);
     }
 
-    @Override
+    /**
+     * cron表达式执行的任务
+     */
     public void scheduleCron(Runnable runnable, String cron) {
         if (SchedulerContext.isStop()) {
             return;

scheduler/src/main/java/com/zfoo/scheduler/manager/SchedulerThreadFactory.java

@@ -1,3 +1,16 @@
+/*
+ * Copyright (C) 2020 The zfoo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and limitations under the License.
+ */
+
 package com.zfoo.scheduler.manager;
 
 import io.netty.util.concurrent.FastThreadLocalThread;

scheduler/src/main/java/com/zfoo/scheduler/schema/SchedulerDefinitionParser.java

@@ -47,8 +47,6 @@ public class SchedulerDefinitionParser implements BeanDefinitionParser {
         builder = BeanDefinitionBuilder.rootBeanDefinition(clazz);
         parserContext.getRegistry().registerBeanDefinition(name, builder.getBeanDefinition());
 
-        // 注册SchedulerManager
-        String schedulerId = element.getAttribute(SCHEDULER_ID);
         return builder.getBeanDefinition();
     }
 

scheduler/src/main/java/com/zfoo/scheduler/schema/SchedulerRegisterProcessor.java

@@ -13,25 +13,70 @@
 
 package com.zfoo.scheduler.schema;
 
-import com.zfoo.scheduler.SchedulerContext;
-import com.zfoo.scheduler.manager.SchedulerManager;
+import com.zfoo.protocol.collection.ArrayUtils;
+import com.zfoo.protocol.util.ReflectionUtils;
+import com.zfoo.protocol.util.StringUtils;
+import com.zfoo.scheduler.manager.SchedulerBus;
+import com.zfoo.scheduler.model.anno.Scheduler;
+import com.zfoo.scheduler.model.vo.SchedulerDefinition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.config.BeanPostProcessor;
 
+import java.lang.reflect.Modifier;
+
 /**
  * @author jaysunxiao
  * @version 3.0
  */
 public class SchedulerRegisterProcessor implements BeanPostProcessor {
 
+    private static final Logger logger = LoggerFactory.getLogger(SchedulerRegisterProcessor.class);
 
     @Override
     public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
-        if (SchedulerContext.getSchedulerContext() == null) {
+        var clazz = bean.getClass();
+        var methods = ReflectionUtils.getMethodsByAnnoInPOJOClass(bean.getClass(), Scheduler.class);
+
+        if (ArrayUtils.isEmpty(methods)) {
             return bean;
         }
-        SchedulerManager schedulerManager = (SchedulerManager) SchedulerContext.getSchedulerManager();
-        schedulerManager.registerScheduler(bean);
+
+        if (!ReflectionUtils.isPojoClass(clazz)) {
+            logger.warn("调度注册类[{}]不是POJO类，父类的调度不会被扫描到", clazz);
+        }
+
+        try {
+            for (var method : methods) {
+                var schedulerMethod = method.getAnnotation(Scheduler.class);
+
+                var paramClazzs = method.getParameterTypes();
+                if (paramClazzs.length >= 1) {
+                    throw new IllegalArgumentException(StringUtils.format("[class:{}] [method:{}] can not have any parameters", bean.getClass(), method.getName()));
+                }
+
+                var methodName = method.getName();
+
+                if (!Modifier.isPublic(method.getModifiers())) {
+                    throw new IllegalArgumentException(StringUtils.format("[class:{}] [method:{}] must use 'public' as modifier!", bean.getClass().getName(), methodName));
+                }
+
+                if (Modifier.isStatic(method.getModifiers())) {
+                    throw new IllegalArgumentException(StringUtils.format("[class:{}] [method:{}] can not use 'static' as modifier!", bean.getClass().getName(), methodName));
+                }
+
+                if (!methodName.startsWith("cron")) {
+                    throw new IllegalArgumentException(StringUtils.format("[class:{}] [method:{}] must start with 'cron' as method name!"
+                            , bean.getClass().getName(), methodName));
+                }
+
+                var scheduler = SchedulerDefinition.valueOf(schedulerMethod.cron(), bean, method);
+                SchedulerBus.registerScheduler(scheduler);
+            }
+        } catch (Throwable t) {
+            throw new RuntimeException(t);
+        }
         return bean;
     }
 

scheduler/src/main/java/com/zfoo/scheduler/util/TimeUtils.java

@@ -13,8 +13,7 @@
 
 package com.zfoo.scheduler.util;
 
-import com.zfoo.scheduler.SchedulerContext;
-import com.zfoo.scheduler.manager.SchedulerManager;
+import com.zfoo.scheduler.manager.SchedulerBus;
 import io.netty.util.concurrent.FastThreadLocal;
 import org.springframework.scheduling.support.CronExpression;
 
@@ -81,8 +80,7 @@ public abstract class TimeUtils {
     };
 
     static {
-        SchedulerContext.getSchedulerManager();
-        SchedulerManager.getInstance();
+        var interval = SchedulerBus.TRIGGER_MILLIS_INTERVAL;
     }
 
     /**