定時任務管理中心（dubbo＋spring）－我們到底能走多遠系列47

我們到底能走多遠系列47

扯淡：

　　又是一年新年時，不知道上一年你付出了多少，收穫了多少呢？也許你正想著老闆會發多少獎金，也許你正想著明年去哪家公司投靠。

　　這個時間點好好整理一下，思考總結一下，的確是個非常好的機會。

　　年終的時候各個公司總會評一下績效，拉出各位的成績單，你是不是想說：去你媽的成績單，我不是你的學生，老子努力工作不是為了看你臉色！當然啦，你想說這話的前提是：你很牛b，如果不是也可以想想，然後默默去變牛b。

　　我大多數的朋友同事都是漂在城市裡的人，我們努力的活得更好，想過自己想過的生活，打心裡佩服我們自己，選擇這個行業，正嘗試改變著世界。

　　所以，加油，各位！

　　另外，程式員過什麼新年？寫bug的時間都不夠呢！

　　最後還是祝看到這個文字的朋友：身體健康，闔家歡樂，雞年大吉，公司上市。

主題：

　　一般，開一個定時任務很簡單，spring寫個註解就能跑了，或者單應用的定時任務還有很多其他豐富jar支持。

public class TaskSupport implements BeanPostProcessor, ApplicationListener<ApplicationContextEvent>,
        ApplicationContextAware {

    private static final Logger logger = LoggerFactory.getLogger(TaskSupport.class);
    private ApplicationContext applicationContext;
    private RegistryConfig registryConfig;
    private ApplicationConfig applicationConfig;
    private ProtocolConfig protocolConfig;
  // 存儲任務bean
    private Map<String, Object> taskBeanMap = new HashMap<String, Object>();
  // dubbo config
    private ServiceConfig<Object> serviceConfig;

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        return bean;
    }
    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        collectTaskBean(bean, beanName);
        return bean;
    }

    private Object getTarget(Object bean) {
        Object target = bean;
        while (target instanceof Advised) {
            try {
                target = ((Advised) bean).getTargetSource().getTarget();
            } catch (Exception e) {
                target = null;
                break;
            }
        }
        return target;
    }

    private void collectTaskBean(Object bean, String beanName) {
        Object target = getTarget(bean);
        if (target != null) {
            Class<?> clazz = target.getClass();
            if (!clazz.isAnnotationPresent(Service.class) || !clazz.isAnnotationPresent(Task.class)) {
                return;
            }
            if (!taskBeanMap.containsKey(beanName)) {
                logger.info("add task bean {}", beanName);
                taskBeanMap.put(beanName, bean);
            }
        }
    }
    @Override
    public void onApplicationEvent(ApplicationContextEvent event) {
        if (isCurrentApplicationContextRefresh(event)) {
            exportTaskDispatcher();
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    /**
     * 向Task暴露任務分發器服務
     */
    protected void exportTaskDispatcher() {
        if (serviceConfig != null && serviceConfig.isExported()) {
            return;
        }
        applicationConfig = applicationContext.getBean(ApplicationConfig.class);
        registryConfig = applicationContext.getBean("soaRegistryConfig", RegistryConfig.class);
        protocolConfig = applicationContext.getBean(ProtocolConfig.class);
        TaskDispatcherImpl taskServiceProxyImpl = wireTaskServiceProxy();
        exportServiceConfig(taskServiceProxyImpl);
    }

    protected void unexportTaskDispatcher() {
        if (serviceConfig != null && serviceConfig.isExported()) {
            serviceConfig.unexport();
        }
    }

    private TaskDispatcherImpl wireTaskServiceProxy() {
        AutowireCapableBeanFactory beanFactory = applicationContext.getAutowireCapableBeanFactory();
        DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) beanFactory;
        TaskDispatcherImpl taskServiceProxy = new TaskDispatcherImpl();
        // ITaskDispatcher的實現bean載入到spring bean容器中，等待dubbo介面暴露
        taskServiceProxy = (TaskDispatcherImpl) beanFactory.initializeBean(taskServiceProxy, "taskServiceProxy");
        // 這裡把篩選出的taskBeanMap註入到TaskDispatcherImpl，直接可以使用
        taskServiceProxy.setTaskBeanMap(taskBeanMap);
        taskServiceProxy.setApplicationConfig(applicationConfig);
        taskServiceProxy.setRegistryConfig(registryConfig);
        defaultListableBeanFactory.registerSingleton("taskServiceProxy", taskServiceProxy);
        return taskServiceProxy;
    }
 /**
     * dubbo介面暴露給任務調度系統
     * 
     */
    private void exportServiceConfig(Object proxy) {
        serviceConfig = new ServiceConfig<Object>();
        serviceConfig.setApplication(applicationConfig);
        serviceConfig.setRegistry(registryConfig);
        serviceConfig.setProtocol(protocolConfig);
        // 把這個介面暴露出去
        serviceConfig.setInterface(ITaskDispatcher.class);
        serviceConfig.setRef(proxy);
        serviceConfig.setRetries(0);
        // 各個業務系統的group不同，這裡充分利用了dubbo的屬性
        serviceConfig.setGroup(applicationConfig.getName());
        serviceConfig.export();
    }

    /**
     * 是否是當前上下文，防止重覆載入和過早載入
     * 
     * @param event
     * @return
     */
    private boolean isCurrentApplicationContextRefresh(ApplicationEvent event) {
        return event instanceof ContextRefreshedEvent
                && ((ContextRefreshedEvent) event).getApplicationContext() == applicationContext;
    }
}

這樣一來，所有應用都會暴露一個ITaskDispatcher 類的方法出去，但是各個group不一樣。ITaskDispatcher定義的方法：

public interface ITaskDispatcher {
   public void dispatch(TaskInvokeInfoDto taskInvokeInfoDto);
}

dispatch方法是調度中心調度觸發啟動任務的方法，根據TaskInvokeInfoDto這個參數里的定義，需要定位到哪一個應用的哪一個類的那一個方法，這個方法的參數是什麼，定位到後執行它，這就是dispatch要實現的功能。

先看一下TaskInvokeInfoDto的定義：

private String appName;//定位到哪個應用，dubbo的group區分
private String beanName;//定位到哪個類
private String methodName;// 定位到哪個方法
private String[] parameterTypes;//方法的參數類型，有重載的情況
private String[] args;//參數值

那麼dispatch的核心代碼：

public void dispatch(TaskInvokeInfoDto taskInvokeInfoDto) {
   try {
      Method method = findMethod(taskInvokeInfoDto);
      Class<?>[] parameterClazzs = method.getParameterTypes();
      if (parameterClazzs.length == 0) {
         ReflectionUtils.invokeMethod(method, taskBeanMap.get(taskInvokeInfoDto.getBeanName()));
      } else {
         Object[] parameterObjs = new Object[parameterClazzs.length];
         for (int i = 0; i < parameterClazzs.length; i++) {
            parameterObjs[i] = Jackson.base().readValue(taskInvokeInfoDto.getArgs()[i], parameterClazzs[i]);
         }
         ReflectionUtils.invokeMethod(method, taskBeanMap.get(taskInvokeInfoDto.getBeanName()), parameterObjs);
      }
    } catch (Exception e) {
      logger.error("execute error...", e);
   }
}
// 上面將的定位邏輯
private Method findMethod(TaskInvokeInfoDto taskInvokeInfoDto) {
        Object bean = taskBeanMap.get(taskInvokeInfoDto.getBeanName());
        Method method = null;
        if (ArrayUtils.isEmpty(taskInvokeInfoDto.getParameterTypes())) {
            method = ReflectionUtils.findMethod(bean.getClass(), taskInvokeInfoDto.getMethodName());
        } else {
            final int paramCount = taskInvokeInfoDto.getParameterTypes().length;
            Class<?>[] clazzArray = new Class<?>[paramCount];
            for (int i = 0; i < paramCount; i++) {
                try {
                    clazzArray[i] = ClassUtils.getClass(taskInvokeInfoDto.getParameterTypes()[i]);
                } catch (ClassNotFoundException e) {
                    logger.info("根據參數類型的字元串創建class對象時失敗", e);
                    return null;
                }
            }
            method = ReflectionUtils.findMethod(bean.getClass(), taskInvokeInfoDto.getMethodName(), clazzArray);
        }
        return method;
    }

以上只要在調度中心處調用dubbo來控制任務執行就可以實現整個任務中心的核心功能。

當然，這裡只是簡單的嘗試性的實現，還有很多優化和擴展可以做，比如任務日誌列印收集，任務應用存活狀態心跳監控，等等。

以前看到過一篇去哪網的吹b文章，吹了半天，仔細看了他提到的功能和沒實現的功能，搞過的人都會覺得做一個其實不難，只是人家分享的時候感覺很厲害，其實他自己心裡清楚自己這個系統也是處處是坑。雖然吹b，不過也會給我們各種啟發。

總結：

1，代碼中利用spring的BeanPostProcessor，篩選出自己需要的bean的方式又是一種新的技巧，我在《請求路由到業務方法設計（2）》中需要篩選bean map用了另一種方式。不知道網友還有其他的想法嗎？

2，反射相關的api還可以繼續深入學習。

讓我們繼續前行

----------------------------------------------------------------------

努力不一定成功，但不努力肯定不會成功。