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

咱們到底能走多遠系列47

扯淡：

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

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

　　年終的時候各個公司總會評一下績效，拉出各位的成績單，你是否是想說：去你媽的成績單，我不是你的學生，老子努力工做不是爲了看你臉色！固然啦，你想說這話的前提是：你很牛b，若是不是也能夠想一想，而後默默去變牛b。

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

　　因此，加油，各位！

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

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

主題：

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

常規的一個場景：

　　一個系統通常都會有不少業務模塊組成，這些業務模塊被封裝成一個個獨立部署的應用拆分出去，獨立維護。各個業務模塊都會有本身定製的定時任務要跑，通常都會依賴本身業務數據和邏輯，很天然的寫在鴿子應用中。

那麼定時任務管理中心要作的是統一管理這些散落在各個業務模塊中的定時任務。

　　

統一管理的好處是：

1，全系統定時任務一目瞭然，便於排查

2，任務執行相關信息統一到一塊兒，好比日誌，而任務業務代碼開發和任務配置解耦 

3，針對任務功能的開發升級集中到一個應用中了

 

這裏粗略設計一個定時任務管理系統拋磚引玉。

大體劃分如下三個部分：

1，任務管理系統

2，任務調度系統

3，業務實現的任務邏輯
任務管理系統用於配置任務的一些信息，任務調度使用這些信息實現對業務系統進行調度，實現定時任務。

拆解後各個組件的關係以下：

 

固然，爲了跟好的描述這個系統，以上圖是一個簡化的設計圖。

如何實現呢？

這裏提供一個代碼的方案，實際開發中結合實際場景和當時技術遺產還有不少的技術方案能夠設計，還能夠深度挖掘。

首先咱們給每一個應用提供任務管理中心的jar包，在業務應用啓動的時候咱們要把任務service收集起來，放入一個map，而後統一提供出一個dubbo接口，用dubbo的group區分各個應用。當任務調度須要調用到這個應用的某個任務service時，再從map中拿出spring bean執行任務方法。

 

以上功能的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還能夠繼續深刻學習。

 

讓咱們繼續前行

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

努力不必定成功，但不努力確定不會成功。