• 较真儿学源码系列-PowerJob启动流程源码分析

            PowerJob版本：4.3.2-main。

    1 简介

            PowerJob是全新一代的分布式任务调度与计算框架，官网地址：http://www.powerjob.tech/。其中介绍了PowerJob的功能特点，以及与其他调度框架的对比，这里就不再赘述了。

            以上是PowerJob的架构图，取自官网。可以看出，PowerJob是典型的客户端/服务端交互的架构（但是在PowerJob中却没有一般分布式中间件会有的注册中心）。本文就从启动流程出发，来一起探究下PowerJob在启动阶段中都做了些什么动作。

    2 服务端

            既然要看启动流程源码，那么首先就来看下启动类。PowerJob依赖于Spring Boot，启动类为PowerJobServerApplication：

    1. /**
    2.  * powerjob-server entry
    3.  *
    4.  * @author tjq
    5.  * @since 2020/3/29
    6.  */
    7. @Slf4j
    8. @EnableScheduling
    9. @SpringBootApplication
    10. public class PowerJobServerApplication {
    11.  
    12.     private static final String TIPS = "\n\n" +
    13.             "******************* PowerJob Tips *******************\n" +
    14.             "如果应用无法启动，我们建议您仔细阅读以下文档来解决:\n" +
    15.             "if server can't startup, we recommend that you read the documentation to find a solution:\n" +
    16.             "https://www.yuque.com/powerjob/guidence/problem\n" +
    17.             "******************* PowerJob Tips *******************\n\n";
    18.  
    19.     public static void main(String[] args) {
    20.  
    21.         pre();
    22.  
    23.         // Start SpringBoot application.
    24.         try {
    25.             SpringApplication.run(PowerJobServerApplication.class, args);
    26.         } catch (Throwable t) {
    27.             log.error(TIPS);
    28.             throw t;
    29.         }
    30.     }
    31.  
    32.     private static void pre() {
    33.         log.info(TIPS);
    34.         PropertyUtils.init();
    35.     }
    36.  
    37. }
    38.  
    39. /**
    40.  * 加载配置文件
    41.  *
    42.  * @author tjq
    43.  * @since 2020/5/18
    44.  */
    45. @Slf4j
    46. public class PropertyUtils {
    47.  
    48.     private static final Properties PROPERTIES = new Properties();
    49.  
    50.     public static Properties getProperties() {
    51.         return PROPERTIES;
    52.     }
    53.  
    54.     public static void init() {
    55.         URL propertiesURL = PropertyUtils.class.getClassLoader().getResource("application.properties");
    56.         Objects.requireNonNull(propertiesURL);
    57.         try (InputStream is = propertiesURL.openStream()) {
    58.             PROPERTIES.load(is);
    59.         } catch (Exception e) {
    60.             ExceptionUtils.rethrow(e);
    61.         }
    62.     }
    63. }

            其中pre方法只是将配置文件内容加载进Properties缓存中。

            既然启动类看不出什么逻辑，那么接下来就来看下服务端启动时有没有什么初始化的逻辑：

    2.1 PowerTransportService

            PowerTransportService类是用作数据传输服务的，也就是客户端和服务端之间的通信。其实现了InitializingBean接口，所以查看下其afterPropertiesSet方法的实现：

    1. /**
    2.  * PowerTransportService:
    3.  */
    4. @Override
    5. public void afterPropertiesSet() throws Exception {
    6.  
    7.     log.info("[PowerTransportService] start to initialize whole PowerTransportService!");
    8.     log.info("[PowerTransportService] activeProtocols: {}", activeProtocols);
    9.  
    10.     if (StringUtils.isEmpty(activeProtocols)) {
    11.         throw new IllegalArgumentException("activeProtocols can't be empty!");
    12.     }
    13.  
    14.     for (String protocol : activeProtocols.split(OmsConstant.COMMA)) {
    15.         try {
    16.             final int port = parseProtocolPort(protocol);
    17.             //初始化网络通讯
    18.             initRemoteFrameWork(protocol, port);
    19.         } catch (Throwable t) {
    20.             log.error("[PowerTransportService] initialize protocol[{}] failed. If you don't need to use this protocol, you can turn it off by 'oms.transporter.active.protocols'", protocol);
    21.             ExceptionUtils.rethrow(t);
    22.         }
    23.     }
    24.  
    25.     //选择默认的通信协议，默认为HTTP
    26.     choseDefault();
    27.  
    28.     log.info("[PowerTransportService] initialize successfully!");
    29.     log.info("[PowerTransportService] ALL_PROTOCOLS: {}", protocolName2Info);
    30. }
    31.  
    32. /**
    33.  * 第18行代码处：
    34.  */
    35. private void initRemoteFrameWork(String protocol, int port) {
    36.  
    37.     // 从构造器注入改为从 applicationContext 获取来避免循环依赖
    38.     //获取所有注解了@Actor的bean
    39.     final Map beansWithAnnotation = applicationContext.getBeansWithAnnotation(Actor.class);
    40.     log.info("[PowerTransportService] find Actor num={},names={}", beansWithAnnotation.size(), beansWithAnnotation.keySet());
    41.  
    42.     Address address = new Address()
    43.             .setHost(NetUtils.getLocalHost())
    44.             .setPort(port);
    45.     EngineConfig engineConfig = new EngineConfig()
    46.             .setServerType(ServerType.SERVER)
    47.             .setType(protocol.toUpperCase())
    48.             .setBindAddress(address)
    49.             .setActorList(Lists.newArrayList(beansWithAnnotation.values()));
    50.     log.info("[PowerTransportService] start to initialize RemoteEngine[type={},address={}]", protocol, address);
    51.     RemoteEngine re = new PowerJobRemoteEngine();
    52.     //初始化网络层
    53.     final EngineOutput engineOutput = re.start(engineConfig);
    54.     log.info("[PowerTransportService] start RemoteEngine[type={},address={}] successfully", protocol, address);
    55.  
    56.     //放入相关缓存中
    57.     this.engines.add(re);
    58.     this.protocolName2Info.put(protocol, new ProtocolInfo(protocol, address.toFullAddress(), engineOutput.getTransporter()));
    59. }

            其中需要说明的一点是：PowerJob网络层使用的协议是Akka或Vert.x：

    • Akka是一个在JVM上构建高并发、分布式和弹性消息驱动的应用程序。其是用Scala写的，使用到了Actor模型；
    • 而Vert.x是一个在JVM上构建响应式应用程序的工具包，其底层基于Netty（之前我对Netty的源码也进行过分析，感兴趣的话可以查看《较真儿学源码系列-Netty核心流程源码分析》）。

            PowerJob屏蔽了底层的实现，用两个自定义的注解@Actor和@Handler进行了统一的封装。客户端传过来的请求会自动跳转到@Actor注解的类、@Handler注解的方法上。拿客户端发送心跳给服务端的逻辑为例，流程如下所示：

            下面就来继续看下，在上面的第53行代码处。PowerJob是如何完成这个绑定的：

    1. /**
    2.  * PowerJobRemoteEngine:
    3.  */
    4. @Override
    5. public EngineOutput start(EngineConfig engineConfig) {
    6.  
    7.     final String engineType = engineConfig.getType();
    8.     EngineOutput engineOutput = new EngineOutput();
    9.     log.info("[PowerJobRemoteEngine] [{}] start remote engine with config: {}", engineType, engineConfig);
    10.     //获取所有@Actor的类，和其中注解了@Handler的方法
    11.     List actorInfos = ActorFactory.load(engineConfig.getActorList());
    12.     //遍历获取指定协议的CSInitializer
    13.     csInitializer = CSInitializerFactory.build(engineType);
    14.  
    15.     String type = csInitializer.type();
    16.  
    17.     Stopwatch sw = Stopwatch.createStarted();
    18.     log.info("[PowerJobRemoteEngine] [{}] try to startup CSInitializer[type={}]", engineType, type);
    19.  
    20.     //CsInitializer初始化，这里以Vert.x为例，查看下其实现
    21.     csInitializer.init(new CSInitializerConfig()
    22.             .setBindAddress(engineConfig.getBindAddress())
    23.             .setServerType(engineConfig.getServerType())
    24.     );
    25.  
    26.     // 构建通讯器
    27.     Transporter transporter = csInitializer.buildTransporter();
    28.     engineOutput.setTransporter(transporter);
    29.  
    30.     log.info("[PowerJobRemoteEngine] [{}] start to bind Handler", engineType);
    31.     actorInfos.forEach(actor -> actor.getHandlerInfos().forEach(handlerInfo -> log.info("[PowerJobRemoteEngine] [{}] PATH={}, handler={}", engineType, handlerInfo.getLocation().toPath(), handlerInfo.getMethod())));
    32.  
    33.     // 绑定 handler
    34.     csInitializer.bindHandlers(actorInfos);
    35.  
    36.     log.info("[PowerJobRemoteEngine] [{}] startup successfully, cost: {}", engineType, sw);
    37.  
    38.     return engineOutput;
    39. }
    40.  
    41. /**
    42.  * ActorFactory:
    43.  * 第11行代码处：
    44.  */
    45. static List load(List actorList) {
    46.  
    47.     List actorInfos = Lists.newArrayList();
    48.  
    49.     actorList.forEach(actor -> {
    50.         final Class clz = actor.getClass();
    51.         try {
    52.             final Actor anno = clz.getAnnotation(Actor.class);
    53.  
    54.             ActorInfo actorInfo = new ActorInfo().setActor(actor).setAnno(anno);
    55.             //获取所有注解了@Handler的方法，并缓存进handlerInfos里
    56.             actorInfo.setHandlerInfos(loadHandlerInfos4Actor(actorInfo));
    57.  
    58.             actorInfos.add(actorInfo);
    59.         } catch (Throwable t) {
    60.             log.error("[ActorFactory] process Actor[{}] failed!", clz);
    61.             ExceptionUtils.rethrow(t);
    62.         }
    63.     });
    64.  
    65.     return actorInfos;
    66. }
    67.  
    68. /**
    69.  * CSInitializerFactory:
    70.  * 第13行代码处：
    71.  */
    72. static CSInitializer build(String targetType) {
    73.  
    74.     Reflections reflections = new Reflections(OmsConstant.PACKAGE);
    75.     //使用Reflections反射获取CSInitializer的实现类，即AkkaCSInitializer和HttpVertxCSInitializer。Reflections的介绍和简单使用请看https://blog.csdn.net/weixin_30342639/article/details/124521467
    76.     Setextends CSInitializer>> cSInitializerClzSet = reflections.getSubTypesOf(CSInitializer.class);
    77.  
    78.     log.info("[CSInitializerFactory] scan subTypeOf CSInitializer: {}", cSInitializerClzSet);
    79.  
    80.     for (Classextends CSInitializer> clz : cSInitializerClzSet) {
    81.         try {
    82.             CSInitializer csInitializer = clz.getDeclaredConstructor().newInstance();
    83.             //获取类型：AKKA/HTTP
    84.             String type = csInitializer.type();
    85.             log.info("[CSInitializerFactory] new instance for CSInitializer[{}] successfully, type={}, object: {}", clz, type, csInitializer);
    86.             //遍历获取指定协议的CSInitializer
    87.             if (targetType.equalsIgnoreCase(type)) {
    88.                 return csInitializer;
    89.             }
    90.         } catch (Exception e) {
    91.             log.error("[CSInitializerFactory] new instance for CSInitializer[{}] failed, maybe you should provide a non-parameter constructor", clz);
    92.             ExceptionUtils.rethrow(e);
    93.         }
    94.     }
    95.  
    96.     throw new PowerJobException(String.format("can't load CSInitializer[%s], ensure your package name start with 'tech.powerjob' and import the dependencies!", targetType));
    97. }
    98.  
    99. /**
    100.  * HttpVertxCSInitializer:
    101.  * 第21行代码处：
    102.  * 这里也就是在做Vert.x的初始化工作，不再继续深入了
    103.  */
    104. @Override
    105. public void init(CSInitializerConfig config) {
    106.     this.config = config;
    107.     vertx = VertxInitializer.buildVertx();
    108.     httpServer = VertxInitializer.buildHttpServer(vertx);
    109.     httpClient = VertxInitializer.buildHttpClient(vertx);
    110. }
    111.  
    112. /**
    113.  * 第34行代码处：
    114.  */
    115. @Override
    116. @SneakyThrows
    117. public void bindHandlers(List actorInfos) {
    118.     Router router = Router.router(vertx);
    119.     // 处理请求响应
    120.     router.route().handler(BodyHandler.create());
    121.     actorInfos.forEach(actorInfo -> {
    122.         Optional.ofNullable(actorInfo.getHandlerInfos()).orElse(Collections.emptyList()).forEach(handlerInfo -> {
    123.             //获取目标地址，即@Actor和@Handler注解上path的拼接
    124.             String handlerHttpPath = handlerInfo.getLocation().toPath();
    125.             ProcessType processType = handlerInfo.getAnno().processType();
    126.  
    127.             Handler routingContextHandler = buildRequestHandler(actorInfo, handlerInfo);
    128.             //添加路由绑定
    129.             Route route = router.post(handlerHttpPath);
    130.             if (processType == ProcessType.BLOCKING) {
    131.                 //绑定阻塞调用handler
    132.                 route.blockingHandler(routingContextHandler, false);
    133.             } else {
    134.                 //绑定非阻塞调用handler
    135.                 route.handler(routingContextHandler);
    136.             }
    137.         });
    138.     });
    139.  
    140.     // 启动 vertx http server
    141.     final int port = config.getBindAddress().getPort();
    142.     final String host = config.getBindAddress().getHost();
    143.  
    144.     httpServer.requestHandler(router)
    145.             .exceptionHandler(e -> log.error("[PowerJob] unknown exception in Actor communication!", e))
    146.             .listen(port, host)
    147.             .toCompletionStage()
    148.             .toCompletableFuture()
    149.             .get(1, TimeUnit.MINUTES);
    150.  
    151.     log.info("[PowerJobRemoteEngine] startup vertx HttpServer successfully!");
    152. }
    153.  
    154. /**
    155.  * 第127行代码处：
    156.  */
    157. private Handler buildRequestHandler(ActorInfo actorInfo, HandlerInfo handlerInfo) {
    158.     Method method = handlerInfo.getMethod();
    159.     Optional> powerSerializeClz = RemoteUtils.findPowerSerialize(method.getParameterTypes());
    160.  
    161.     // 内部框架，严格模式，绑定失败直接报错
    162.     if (!powerSerializeClz.isPresent()) {
    163.         throw new PowerJobException("can't find any 'PowerSerialize' object in handler args: " + handlerInfo.getLocation());
    164.     }
    165.  
    166.     //这里实际上是注册了一个事件驱动的回调函数（Netty的玩法），当有请求过来的时候，会走到下面的代码里
    167.     return ctx -> {
    168.         final RequestBody body = ctx.body();
    169.         final Object convertResult = body.asPojo(powerSerializeClz.get());
    170.         try {
    171.             //这里通过反射调用相关的@Handler注解的方法
    172.             Object response = method.invoke(actorInfo.getActor(), convertResult);
    173.             if (response != null) {
    174.                 if (response instanceof String) {
    175.                     ctx.end((String) response);
    176.                 } else {
    177.                     ctx.json(JsonObject.mapFrom(response));
    178.                 }
    179.                 return;
    180.             }
    181.  
    182.             ctx.end();
    183.         } catch (Throwable t) {
    184.             // 注意这里是框架实际运行时，日志输出用标准 PowerJob 格式
    185.             log.error("[PowerJob] invoke Handler[{}] failed!", handlerInfo.getLocation(), t);
    186.             ctx.fail(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), t);
    187.         }
    188.     };
    189. }
      190.  

              通过上面的buildRequestHandler方法可知，当Vert.x接收到客户端的请求时，会调用到一个回调函数。而这个回调函数会最终通过反射的方式调用到自定义的@Handler注解的方法中来。
       

              上面只是完成了初始化和绑定的操作，还缺少具体调用时候的逻辑（这里属于一个整体流程，所以就一起分析了）。继续拿发送心跳为例，调用的方法是：TransportUtils.reportWorkerHeartbeat：
       

      1. /**
      2.  * TransportUtils:
      3.  */
      4. public static void reportWorkerHeartbeat(WorkerHeartbeat req, String address, Transporter transporter) {
      5.     //绑定url调用信息
      6.     final URL url = easyBuildUrl(ServerType.SERVER, S4W_PATH, S4W_HANDLER_WORKER_HEARTBEAT, address);
      7.     //这里依旧是拿Vert.x的实现来分析
      8.     transporter.tell(url, req);
      9. }
      10.  
      11. /**
      12.  * VertxTransporter:
      13.  * 第8行代码处：
      14.  */
      15. @Override
      16. public void tell(URL url, PowerSerializable request) {
      17.     post(url, request, null);
      18. }
      19.  
      20. /**
      21.  * 这里就涉及到Vert.x的具体调用细节了，不再深入
      22.  */
      23. @SuppressWarnings("unchecked")
      24. private  CompletionStage post(URL url, PowerSerializable request, Class clz) {
      25.     final String host = url.getAddress().getHost();
      26.     final int port = url.getAddress().getPort();
      27.     final String path = url.getLocation().toPath();
      28.     RequestOptions requestOptions = new RequestOptions()
      29.             .setMethod(HttpMethod.POST)
      30.             .setHost(host)
      31.             .setPort(port)
      32.             .setURI(path);
      33.     // 获取远程服务器的HTTP连接
      34.     Future httpClientRequestFuture = httpClient.request(requestOptions);
      35.     // 转换 -> 发送请求获取响应
      36.     Future responseFuture = httpClientRequestFuture.compose(httpClientRequest ->
      37.             httpClientRequest
      38.                     .putHeader(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON)
      39.                     .send(JsonObject.mapFrom(request).toBuffer())
      40.     );
      41.     return responseFuture.compose(httpClientResponse -> {
      42.         // throw exception
      43.         final int statusCode = httpClientResponse.statusCode();
      44.         if (statusCode != HttpResponseStatus.OK.code()) {
      45.             // CompletableFuture.get() 时会传递抛出该异常
      46.             throw new RemotingException(String.format("request [host:%s,port:%s,url:%s] failed, status: %d, msg: %s",
      47.                     host, port, path, statusCode, httpClientResponse.statusMessage()
      48.             ));
      49.         }
      50.  
      51.         return httpClientResponse.body().compose(x -> {
      52.  
      53.             if (clz == null) {
      54.                 return Future.succeededFuture(null);
      55.             }
      56.  
      57.             if (clz.equals(String.class)) {
      58.                 return Future.succeededFuture((T) x.toString());
      59.             }
      60.  
      61.             return Future.succeededFuture(x.toJsonObject().mapTo(clz));
      62.         });
      63.     }).toCompletionStage();
      64. }
       

              当客户端往服务端发送完请求后，服务端接受到相关的请求，会调用到相应的回调函数，从而最终调用到@Handler注解的方法。整个流程就串起来了。
       

      2.2. InstanceMetadataService
       

              同PowerTransportService类一样，InstanceMetadataService也实现了InitializingBean接口，所以查看下其afterPropertiesSet方法的实现：
       

      1. /**
      2.  * InstanceMetadataService:
      3.  */
      4. @Override
      5. public void afterPropertiesSet() throws Exception {
      6.     instanceId2JobInfoCache = CacheBuilder.newBuilder()
      7.             .concurrencyLevel(CACHE_CONCURRENCY_LEVEL)
      8.             .maximumSize(instanceMetadataCacheSize)
      9.             .softValues()
      10.             .build();
      11. }
       

              其中只是初始化了一个本地缓存，没有多余的逻辑。
       

      2.3 CoreScheduleTaskManager
       

              CoreScheduleTaskManager也实现了InitializingBean接口，查看下其afterPropertiesSet方法的实现：
       

      1. /**
      2.  * CoreScheduleTaskManager:
      3.  */
      4. @SuppressWarnings("AlibabaAvoidManuallyCreateThread")
      5. @Override
      6. public void afterPropertiesSet() {
      7.     // 定时调度
      8.     coreThreadContainer.add(new Thread(new LoopRunnable("ScheduleCronJob", PowerScheduleService.SCHEDULE_RATE, () -> powerScheduleService.scheduleNormalJob(TimeExpressionType.CRON)), "Thread-ScheduleCronJob"));
      9.     coreThreadContainer.add(new Thread(new LoopRunnable("ScheduleDailyTimeIntervalJob", PowerScheduleService.SCHEDULE_RATE, () -> powerScheduleService.scheduleNormalJob(TimeExpressionType.DAILY_TIME_INTERVAL)), "Thread-ScheduleDailyTimeIntervalJob"));
      10.     coreThreadContainer.add(new Thread(new LoopRunnable("ScheduleCronWorkflow", PowerScheduleService.SCHEDULE_RATE, powerScheduleService::scheduleCronWorkflow), "Thread-ScheduleCronWorkflow"));
      11.     coreThreadContainer.add(new Thread(new LoopRunnable("ScheduleFrequentJob", PowerScheduleService.SCHEDULE_RATE, powerScheduleService::scheduleFrequentJob), "Thread-ScheduleFrequentJob"));
      12.     // 数据清理
      13.     coreThreadContainer.add(new Thread(new LoopRunnable("CleanWorkerData", PowerScheduleService.SCHEDULE_RATE, powerScheduleService::cleanData), "Thread-CleanWorkerData"));
      14.     // 状态检查
      15.     coreThreadContainer.add(new Thread(new LoopRunnable("CheckRunningInstance", InstanceStatusCheckService.CHECK_INTERVAL, instanceStatusCheckService::checkRunningInstance), "Thread-CheckRunningInstance"));
      16.     coreThreadContainer.add(new Thread(new LoopRunnable("CheckWaitingDispatchInstance", InstanceStatusCheckService.CHECK_INTERVAL, instanceStatusCheckService::checkWaitingDispatchInstance), "Thread-CheckWaitingDispatchInstance"));
      17.     coreThreadContainer.add(new Thread(new LoopRunnable("CheckWaitingWorkerReceiveInstance", InstanceStatusCheckService.CHECK_INTERVAL, instanceStatusCheckService::checkWaitingWorkerReceiveInstance), "Thread-CheckWaitingWorkerReceiveInstance"));
      18.     coreThreadContainer.add(new Thread(new LoopRunnable("CheckWorkflowInstance", InstanceStatusCheckService.CHECK_INTERVAL, instanceStatusCheckService::checkWorkflowInstance), "Thread-CheckWorkflowInstance"));
      19.  
      20.     coreThreadContainer.forEach(Thread::start);
      21. }
       

              可以看到，其中初始化了一堆的定时任务。从中挑几个定时任务来分析下：
       

      2.3.1 ScheduleCronJob/ScheduleDailyTimeIntervalJob
       

      1. /**
      2.  * PowerScheduleService:
      3.  */
      4. public void scheduleNormalJob(TimeExpressionType timeExpressionType) {
      5.     long start = System.currentTimeMillis();
      6.     // 调度 CRON 表达式 JOB
      7.     try {
      8.         //获取在PowerJob控制台配置的appId，也就是服务id（PowerJob会使用数据库来存储服务、实例等相关数据。在进行一些操作前，会先落库，然后再执行。以此避免相关操作丢失的情况出现）
      9.         final List allAppIds = appInfoRepository.listAppIdByCurrentServer(transportService.defaultProtocol().getAddress());
      10.         if (CollectionUtils.isEmpty(allAppIds)) {
      11.             log.info("[NormalScheduler] current server has no app's job to schedule.");
      12.             return;
      13.         }
      14.         scheduleNormalJob0(timeExpressionType, allAppIds);
      15.     } catch (Exception e) {
      16.         log.error("[NormalScheduler] schedule cron job failed.", e);
      17.     }
      18.     long cost = System.currentTimeMillis() - start;
      19.     log.info("[NormalScheduler] {} job schedule use {} ms.", timeExpressionType, cost);
      20.     if (cost > SCHEDULE_RATE) {
      21.         log.warn("[NormalScheduler] The database query is using too much time({}ms), please check if the database load is too high!", cost);
      22.     }
      23. }
      24.  
      25. /**
      26.  * 第14行代码处：
      27.  * 调度普通服务端计算表达式类型（CRON、DAILY_TIME_INTERVAL）的任务
      28.  *
      29.  * @param timeExpressionType 表达式类型
      30.  * @param appIds             appIds
      31.  */
      32. private void scheduleNormalJob0(TimeExpressionType timeExpressionType, List appIds) {
      33.  
      34.     long nowTime = System.currentTimeMillis();
      35.     long timeThreshold = nowTime + 2 * SCHEDULE_RATE;
      36.     //分组执行
      37.     Lists.partition(appIds, MAX_APP_NUM).forEach(partAppIds -> {
      38.  
      39.         try {
      40.  
      41.             // 查询条件：任务开启 + 使用CRON表达调度时间 + 指定appId + 即将需要调度执行
      42.             //获取在PowerJob控制台配置的执行任务
      43.             List jobInfos = jobInfoRepository.findByAppIdInAndStatusAndTimeExpressionTypeAndNextTriggerTimeLessThanEqual(partAppIds, SwitchableStatus.ENABLE.getV(), timeExpressionType.getV(), timeThreshold);
      44.  
      45.             if (CollectionUtils.isEmpty(jobInfos)) {
      46.                 return;
      47.             }
      48.  
      49.             // 1. 批量写日志表
      50.             Map jobId2InstanceId = Maps.newHashMap();
      51.             log.info("[NormalScheduler] These {} jobs will be scheduled: {}.", timeExpressionType.name(), jobInfos);
      52.  
      53.             jobInfos.forEach(jobInfo -> {
      54.                 //实例表进行落库（任务和实例的关系为：每执行一次任务会生成一个实例）
      55.                 Long instanceId = instanceService.create(jobInfo.getId(), jobInfo.getAppId(), jobInfo.getJobParams(), null, null, jobInfo.getNextTriggerTime()).getInstanceId();
      56.                 jobId2InstanceId.put(jobInfo.getId(), instanceId);
      57.             });
      58.             instanceInfoRepository.flush();
      59.  
      60.             // 2. 推入时间轮中等待调度执行（对时间轮代码的分析请查看https://blog.csdn.net/weixin_30342639/article/details/132732836）
      61.             jobInfos.forEach(jobInfoDO -> {
      62.  
      63.                 Long instanceId = jobId2InstanceId.get(jobInfoDO.getId());
      64.  
      65.                 //获取下次执行时间
      66.                 long targetTriggerTime = jobInfoDO.getNextTriggerTime();
      67.                 long delay = 0;
      68.                 if (targetTriggerTime < nowTime) {
      69.                     log.warn("[Job-{}] schedule delay, expect: {}, current: {}", jobInfoDO.getId(), targetTriggerTime, System.currentTimeMillis());
      70.                 } else {
      71.                     //计算距离下次执行时间所需要延迟的时间
      72.                     delay = targetTriggerTime - nowTime;
      73.                 }
      74.  
      75.                 //任务实例推入时间轮，等待被执行
      76.                 InstanceTimeWheelService.schedule(instanceId, delay, () -> dispatchService.dispatch(jobInfoDO, instanceId, Optional.empty(), Optional.empty()));
      77.             });
      78.  
      79.             // 3. 计算下一次调度时间（忽略5S内的重复执行，即CRON模式下最小的连续执行间隔为 SCHEDULE_RATE ms）
      80.             jobInfos.forEach(jobInfoDO -> {
      81.                 try {
      82.                     //重新计算下次执行时间，并落库
      83.                     //这里会用到策略模式，不同类型的任务（Cron/固定频率/每日固定间隔）会有不同的计算方式，具体不再深入分析，感兴趣可自行查看
      84.                     refreshJob(timeExpressionType, jobInfoDO);
      85.                 } catch (Exception e) {
      86.                     log.error("[Job-{}] refresh job failed.", jobInfoDO.getId(), e);
      87.                 }
      88.             });
      89.             jobInfoRepository.flush();
      90.  
      91.  
      92.         } catch (Exception e) {
      93.             log.error("[NormalScheduler] schedule {} job failed.", timeExpressionType.name(), e);
      94.         }
      95.     });
      96. }
       

              在上面第76行代码处，当任务推入到时间轮之后（对时间轮代码的分析请查看《较真儿学源码系列-PowerJob时间轮源码分析》），等到需要被执行的时候，会调用到DispatchService.dispatch方法来派发任务：
       

      1. /**
      2.  * DispatchService:
      3.  * 将任务从Server派发到Worker（TaskTracker）
      4.  * 只会派发当前状态为等待派发的任务实例
      5.  * **************************************************
      6.  * 2021-02-03 modify by Echo009
      7.  * 1、移除参数 当前运行次数、工作流实例ID、实例参数
      8.  * 更改为从当前任务实例中获取获取以上信息
      9.  * 2、移除运行次数相关的（runningTimes）处理逻辑
      10.  * 迁移至 {@link InstanceManager#updateStatus} 中处理
      11.  * **************************************************
      12.  *
      13.  * @param jobInfo              任务的元信息
      14.  * @param instanceId           任务实例ID
      15.  * @param instanceInfoOptional 任务实例信息，可选
      16.  * @param overloadOptional     超载信息，可选
      17.  */
      18. @UseCacheLock(type = "processJobInstance", key = "#jobInfo.getMaxInstanceNum() > 0 || T(tech.powerjob.common.enums.TimeExpressionType).FREQUENT_TYPES.contains(#jobInfo.getTimeExpressionType()) ? #jobInfo.getId() : #instanceId", concurrencyLevel = 1024)
      19. public void dispatch(JobInfoDO jobInfo, Long instanceId, Optional instanceInfoOptional, Optional> overloadOptional) {
      20.     // 允许从外部传入实例信息，减少 io 次数
      21.     // 检查当前任务是否被取消
      22.     //获取实例
      23.     InstanceInfoDO instanceInfo = instanceInfoOptional.orElseGet(() -> instanceInfoRepository.findByInstanceId(instanceId));
      24.     Long jobId = instanceInfo.getJobId();
      25.         
      26.     //...
      27.  
      28.     // 任务信息已经被删除
      29.     if (jobInfo.getId() == null) {
      30.         log.warn("[Dispatcher-{}|{}] cancel dispatch due to job(id={}) has been deleted!", jobId, instanceId, jobId);
      31.         instanceManager.processFinishedInstance(instanceId, instanceInfo.getWfInstanceId(), FAILED, "can't find job by id " + jobId);
      32.         return;
      33.     }
      34.  
      35.     Date now = new Date();
      36.     String dbInstanceParams = instanceInfo.getInstanceParams() == null ? "" : instanceInfo.getInstanceParams();
      37.     log.info("[Dispatcher-{}|{}] start to dispatch job: {};instancePrams: {}.", jobId, instanceId, jobInfo, dbInstanceParams);
      38.  
      39.     // 查询当前运行的实例数
      40.     long current = System.currentTimeMillis();
      41.     Integer maxInstanceNum = jobInfo.getMaxInstanceNum();
      42.     // 秒级任务只派发到一台机器，具体的 maxInstanceNum 由 TaskTracker 控制
      43.     if (TimeExpressionType.FREQUENT_TYPES.contains(jobInfo.getTimeExpressionType())) {
      44.         maxInstanceNum = 1;
      45.     }
      46.  
      47.     //...
      48.         
      49.     // 获取当前最合适的 worker 列表
      50.     List suitableWorkers = workerClusterQueryService.getSuitableWorkers(jobInfo);
      51.  
      52.     //...
      53.  
      54.     // 判断是否超载，在所有可用 worker 超载的情况下直接跳过当前任务
      55.     suitableWorkers = filterOverloadWorker(suitableWorkers);
      56.         
      57.     //...
      58.  
      59.     //获取worker ip
      60.     List workerIpList = suitableWorkers.stream().map(WorkerInfo::getAddress).collect(Collectors.toList());
      61.     // 构造任务调度请求
      62.     ServerScheduleJobReq req = constructServerScheduleJobReq(jobInfo, instanceInfo, workerIpList);
      63.  
      64.     // 发送请求（不可靠，需要一个后台线程定期轮询状态）
      65.     //只取第一个worker
      66.     WorkerInfo taskTracker = suitableWorkers.get(0);
      67.     String taskTrackerAddress = taskTracker.getAddress();
      68.  
      69.     URL workerUrl = ServerURLFactory.dispatchJob2Worker(taskTrackerAddress);
      70.     transportService.tell(taskTracker.getProtocol(), workerUrl, req);
      71.     log.info("[Dispatcher-{}|{}] send schedule request to TaskTracker[protocol:{},address:{}] successfully: {}.", jobId, instanceId, taskTracker.getProtocol(), taskTrackerAddress, req);
      72.  
      73.     // 修改状态
      74.     //修改实例表状态
      75.     instanceInfoRepository.update4TriggerSucceed(instanceId, WAITING_WORKER_RECEIVE.getV(), current, taskTrackerAddress, now, instanceInfo.getStatus());
      76.     // 装载缓存
      77.     instanceMetadataService.loadJobInfo(instanceId, jobInfo);
      78. }
      79.  
      80. /**
      81.  * InstanceManager:
      82.  * 第31行代码处：
      83.  * 收尾完成的任务实例
      84.  *
      85.  * @param instanceId   任务实例ID
      86.  * @param wfInstanceId 工作流实例ID，非必须
      87.  * @param status       任务状态，有 成功/失败/手动停止
      88.  * @param result       执行结果
      89.  */
      90. public void processFinishedInstance(Long instanceId, Long wfInstanceId, InstanceStatus status, String result) {
      91.  
      92.     log.info("[Instance-{}] process finished, final status is {}.", instanceId, status.name());
      93.  
      94.     // 上报日志数据
      95.     //时间轮延迟执行
      96.     HashedWheelTimerHolder.INACCURATE_TIMER.schedule(() -> instanceLogService.sync(instanceId), 60, TimeUnit.SECONDS);
      97.  
      98.     // workflow 特殊处理
      99.     if (wfInstanceId != null) {
      100.         // 手动停止在工作流中也认为是失败（理论上不应该发生）
      101.         workflowInstanceManager.move(wfInstanceId, instanceId, status, result);
      102.     }
      103.  
      104.     // 告警
      105.     if (status == InstanceStatus.FAILED) {
      106.         alert(instanceId, result);
      107.     }
      108.     // 主动移除缓存，减小内存占用
      109.     instanceMetadataService.invalidateJobInfo(instanceId);
      110. }
      111.  
      112. /**
      113.  * InstanceLogService:
      114.  * 第96行代码处：
      115.  * 将本地的任务实例运行日志同步到 mongoDB 存储，在任务执行结束后异步执行
      116.  *
      117.  * @param instanceId 任务实例ID
      118.  */
      119. @Async(PJThreadPool.BACKGROUND_POOL)
      120. public void sync(Long instanceId) {
      121.  
      122.     Stopwatch sw = Stopwatch.createStarted();
      123.     try {
      124.         // 先持久化到本地文件
      125.         File stableLogFile = genStableLogFile(instanceId);
      126.         // 将文件推送到 MongoDB
      127.         if (gridFsManager.available()) {
      128.             try {
      129.                 gridFsManager.store(stableLogFile, GridFsManager.LOG_BUCKET, genMongoFileName(instanceId));
      130.                 log.info("[InstanceLog-{}] push local instanceLogs to mongoDB succeed, using: {}.", instanceId, sw.stop());
      131.             } catch (Exception e) {
      132.                 log.warn("[InstanceLog-{}] push local instanceLogs to mongoDB failed.", instanceId, e);
      133.             }
      134.         }
      135.     } catch (Exception e) {
      136.         log.warn("[InstanceLog-{}] sync local instanceLogs failed.", instanceId, e);
      137.     }
      138.     // 删除本地数据库数据
      139.     try {
      140.         instanceId2LastReportTime.remove(instanceId);
      141.         CommonUtils.executeWithRetry0(() -> localInstanceLogRepository.deleteByInstanceId(instanceId));
      142.         log.info("[InstanceLog-{}] delete local instanceLog successfully.", instanceId);
      143.     } catch (Exception e) {
      144.         log.warn("[InstanceLog-{}] delete local instanceLog failed.", instanceId, e);
      145.     }
      146. }
      147.  
      148. /**
      149.  * WorkerClusterQueryService:
      150.  * 第50行代码处：
      151.  * get worker for job
      152.  *
      153.  * @param jobInfo job
      154.  * @return worker cluster info, sorted by metrics desc
      155.  */
      156. public List getSuitableWorkers(JobInfoDO jobInfo) {
      157.  
      158.     //获取该集群所有的机器信息
      159.     List workers = Lists.newLinkedList(getWorkerInfosByAppId(jobInfo.getAppId()).values());
      160.  
      161.     workers.removeIf(workerInfo -> filterWorker(workerInfo, jobInfo));
      162.  
      163.     DispatchStrategy dispatchStrategy = DispatchStrategy.of(jobInfo.getDispatchStrategy());
      164.     switch (dispatchStrategy) {
      165.         case RANDOM:
      166.             //随机的方式就是打乱顺序
      167.             Collections.shuffle(workers);
      168.             break;
      169.         case HEALTH_FIRST:
      170.             //健康优先的方式需要计算下得分，按得分高低排序
      171.             workers.sort((o1, o2) -> o2.getSystemMetrics().calculateScore() - o1.getSystemMetrics().calculateScore());
      172.             break;
      173.         default:
      174.             // do nothing
      175.     }
      176.  
      177.     // 限定集群大小（0代表不限制）
      178.     if (!workers.isEmpty() && jobInfo.getMaxWorkerCount() > 0 && workers.size() > jobInfo.getMaxWorkerCount()) {
      179.         workers = workers.subList(0, jobInfo.getMaxWorkerCount());
      180.     }
      181.     return workers;
      182. }
      183.  
      184. /**
      185.  * SystemMetrics:
      186.  * 第171行代码处：
      187.  * Calculate score, based on CPU and memory info.
      188.  *
      189.  * @return score
      190.  */
      191. public int calculateScore() {
      192.     if (score > 0) {
      193.         return score;
      194.     }
      195.     // Memory is vital to TaskTracker, so we set the multiplier factor as 2.
      196.     //未使用内存指标的权重为2
      197.     double memScore = (jvmMaxMemory - jvmUsedMemory) * 2;
      198.     // Calculate the remaining load of CPU. Multiplier is set as 1.
      199.     //剩余可用cpu数的权重为1
      200.     double cpuScore = cpuProcessors - cpuLoad;
      201.     // Windows can not fetch CPU load, set cpuScore as 1.
      202.     //Windows系统拿不到cpu使用数，所以cpu得分设置为1
      203.     if (cpuScore > cpuProcessors) {
      204.         cpuScore = 1;
      205.     }
      206.     //最终得分就是内存得分+cpu得分
      207.     score = (int) (memScore + cpuScore);
      208.     return score;
      209. }
      210.  
      211. /**
      212.  * ServerURLFactory:
      213.  * 第69行代码处：
      214.  */
      215. public static URL dispatchJob2Worker(String address) {
      216.     return simileBuild(address, ServerType.WORKER, WORKER_PATH, WTT_HANDLER_RUN_JOB);
      217. }
       

              在上面第216行代码处会将请求跳转到WorkerActor.onReceiveServerScheduleJobReq方法：
       

      1. /**
      2.  * WorkerActor:
      3.  */
      4. @Handler(path = WTT_HANDLER_RUN_JOB)
      5. public void onReceiveServerScheduleJobReq(ServerScheduleJobReq req) {
      6.     taskTrackerActor.onReceiveServerScheduleJobReq(req);
      7. }
      8.  
      9. /**
      10.  * TaskTrackerActor:
      11.  * 服务器任务调度处理器
      12.  */
      13. @Handler(path = WTT_HANDLER_RUN_JOB)
      14. public void onReceiveServerScheduleJobReq(ServerScheduleJobReq req) {
      15.     log.debug("[TaskTrackerActor] server schedule job by request: {}.", req);
      16.     Long instanceId = req.getInstanceId();
      17.     // 区分轻量级任务模型以及重量级任务模型
      18.     //单机执行的OpenApi/Corn/工作流是轻量级任务，其他的是重量级任务
      19.     if (isLightweightTask(req)) {
      20.         final LightTaskTracker taskTracker = LightTaskTrackerManager.getTaskTracker(instanceId);
      21.         if (taskTracker != null) {
      22.             log.warn("[TaskTrackerActor] LightTaskTracker({}) for instance(id={}) already exists.", taskTracker, instanceId);
      23.             return;
      24.         }
      25.         // 判断是否已经 overload
      26.         if (LightTaskTrackerManager.currentTaskTrackerSize() >= workerRuntime.getWorkerConfig().getMaxLightweightTaskNum() * LightTaskTrackerManager.OVERLOAD_FACTOR) {
      27.             // ignore this request
      28.             log.warn("[TaskTrackerActor] this worker is overload,ignore this request(instanceId={}),current size = {}!", instanceId, LightTaskTrackerManager.currentTaskTrackerSize());
      29.             return;
      30.         }
      31.         if (LightTaskTrackerManager.currentTaskTrackerSize() >= workerRuntime.getWorkerConfig().getMaxLightweightTaskNum()) {
      32.             log.warn("[TaskTrackerActor] this worker will be overload soon,current size = {}!", LightTaskTrackerManager.currentTaskTrackerSize());
      33.         }
      34.         // 创建轻量级任务
      35.         //在轻量级本地缓存中添加TaskTracker
      36.         LightTaskTrackerManager.atomicCreateTaskTracker(instanceId, ignore -> LightTaskTracker.create(req, workerRuntime));
      37.     } else {
      38.         HeavyTaskTracker taskTracker = HeavyTaskTrackerManager.getTaskTracker(instanceId);
      39.         if (taskTracker != null) {
      40.             log.warn("[TaskTrackerActor] HeavyTaskTracker({}) for instance(id={}) already exists.", taskTracker, instanceId);
      41.             return;
      42.         }
      43.         // 判断是否已经 overload
      44.         if (HeavyTaskTrackerManager.currentTaskTrackerSize() >= workerRuntime.getWorkerConfig().getMaxHeavyweightTaskNum()) {
      45.             // ignore this request
      46.             log.warn("[TaskTrackerActor] this worker is overload,ignore this request(instanceId={})! current size = {},", instanceId, HeavyTaskTrackerManager.currentTaskTrackerSize());
      47.             return;
      48.         }
      49.         // 原子创建，防止多实例的存在
      50.         //在重量级本地缓存中添加TaskTracker
      51.         HeavyTaskTrackerManager.atomicCreateTaskTracker(instanceId, ignore -> HeavyTaskTracker.create(req, workerRuntime));
      52.     }
      53. }
       

              由上可以看到，派发任务就是往TaskTracker里添加一条本地缓存。
       

              这里出现了TaskTracker，TaskTracker、ProcessorTracker和Processor的概念直接参考PowerJob作者写过的文章：
       

       

              这里我们继续分析下上面第51行代码处的HeavyTaskTracker.create方法：
       

      1. /**
      2.  * HeavyTaskTracker:
      3.  * 静态方法创建 TaskTracker
      4.  *
      5.  * @param req 服务端调度任务请求
      6.  * @return API/CRON -> CommonTaskTracker, FIX_RATE/FIX_DELAY -> FrequentTaskTracker
      7.  */
      8. public static HeavyTaskTracker create(ServerScheduleJobReq req, WorkerRuntime workerRuntime) {
      9.     try {
      10.         TimeExpressionType timeExpressionType = TimeExpressionType.valueOf(req.getTimeExpressionType());
      11.         switch (timeExpressionType) {
      12.             case FIXED_RATE:
      13.             case FIXED_DELAY:
      14.                 return new FrequentTaskTracker(req, workerRuntime);
      15.             default:
      16.                 //这里我们分析下CommonTaskTracker构造器的实现
      17.                 return new CommonTaskTracker(req, workerRuntime);
      18.         }
      19.     } catch (Exception e) {
      20.         reportCreateErrorToServer(req, workerRuntime, e);
      21.     }
      22.     return null;
      23. }
      24.  
      25. /**
      26.  * CommonTaskTracker:
      27.  * 第17行代码处：
      28.  */
      29. protected CommonTaskTracker(ServerScheduleJobReq req, WorkerRuntime workerRuntime) {
      30.     super(req, workerRuntime);
      31. }
      32.  
      33. /**
      34.  * HeavyTaskTracker:
      35.  */
      36. protected HeavyTaskTracker(ServerScheduleJobReq req, WorkerRuntime workerRuntime) {
      37.     // 初始化成员变量
      38.     super(req, workerRuntime);
      39.     // 赋予时间表达式类型
      40.     instanceInfo.setTimeExpressionType(TimeExpressionType.valueOf(req.getTimeExpressionType()).getV());
      41.     // 保护性操作
      42.     instanceInfo.setThreadConcurrency(Math.max(1, instanceInfo.getThreadConcurrency()));
      43.     this.ptStatusHolder = new ProcessorTrackerStatusHolder(instanceId, req.getMaxWorkerCount(), req.getAllWorkerAddress());
      44.     this.taskPersistenceService = workerRuntime.getTaskPersistenceService();
      45.     // 构建缓存
      46.     taskId2BriefInfo = CacheBuilder.newBuilder().maximumSize(1024).build();
      47.  
      48.     // 构建分段锁
      49.     //SegmentLock是自己实现的ReentrantLock数组（使用数组是为了提高并发度）
      50.     segmentLock = new SegmentLock(UPDATE_CONCURRENCY);
      51.  
      52.     // 子类自定义初始化操作
      53.     initTaskTracker(req);
      54.  
      55.     log.info("[TaskTracker-{}] create TaskTracker successfully.", instanceId);
      56. }
      57.  
      58. /**
      59.  * CommonTaskTracker:
      60.  * 第53行代码处：
      61.  *
      62.  * @param req 服务器调度任务实例运行请求
      63.  */
      64. @Override
      65. protected void initTaskTracker(ServerScheduleJobReq req) {
      66.  
      67.     // CommonTaskTrackerTimingPool 缩写
      68.     String poolName = String.format("ctttp-%d", req.getInstanceId()) + "-%d";
      69.     ThreadFactory factory = new ThreadFactoryBuilder().setNameFormat(poolName).build();
      70.     this.scheduledPool = Executors.newScheduledThreadPool(2, factory);
      71.  
      72.     // 持久化根任务
      73.     persistenceRootTask();
      74.  
      75.     // 开启定时状态检查
      76.     int delay = Integer.parseInt(System.getProperty(PowerJobDKey.WORKER_STATUS_CHECK_PERIOD, "13"));
      77.     scheduledPool.scheduleWithFixedDelay(new StatusCheckRunnable(), 3, delay, TimeUnit.SECONDS);
      78.  
      79.     // 如果是 MR 任务，则需要启动执行器动态检测装置
      80.     ExecuteType executeType = ExecuteType.valueOf(req.getExecuteType());
      81.     if (executeType == ExecuteType.MAP || executeType == ExecuteType.MAP_REDUCE) {
      82.         scheduledPool.scheduleAtFixedRate(new WorkerDetector(), 1, 1, TimeUnit.MINUTES);
      83.     }
      84.  
      85.     // 最后启动任务派发器，否则会出现 TaskTracker 还未创建完毕 ProcessorTracker 已开始汇报状态的情况
      86.     scheduledPool.scheduleWithFixedDelay(new Dispatcher(), 10, 5000, TimeUnit.MILLISECONDS);
      87. }
      88.  
      89. /**
      90.  * 第73行代码处：
      91.  * 持久化根任务，只有完成持久化才能视为任务开始running（先持久化，再报告server）
      92.  */
      93. private void persistenceRootTask() {
      94.  
      95.     TaskDO rootTask = new TaskDO();
      96.     rootTask.setStatus(TaskStatus.WAITING_DISPATCH.getValue());
      97.     rootTask.setInstanceId(instanceInfo.getInstanceId());
      98.     rootTask.setTaskId(ROOT_TASK_ID);
      99.     rootTask.setFailedCnt(0);
      100.     rootTask.setAddress(workerRuntime.getWorkerAddress());
      101.     //这里需要留意下，根任务的名称为OMS_ROOT_TASK（ROOT_TASK_NAME）
      102.     rootTask.setTaskName(TaskConstant.ROOT_TASK_NAME);
      103.     rootTask.setCreatedTime(System.currentTimeMillis());
      104.     rootTask.setLastModifiedTime(System.currentTimeMillis());
      105.     rootTask.setLastReportTime(-1L);
      106.     rootTask.setSubInstanceId(instanceId);
      107.  
      108.     if (taskPersistenceService.save(rootTask)) {
      109.         log.info("[TaskTracker-{}] create root task successfully.", instanceId);
      110.     } else {
      111.         log.error("[TaskTracker-{}] create root task failed.", instanceId);
      112.         throw new PowerJobException("create root task failed for instance: " + instanceId);
      113.     }
      114. }
       

              在上面第77行代码处、第82行代码处和第86行代码处的initTaskTracker方法中，分别开启了三个定时任务，这里我们主要分析下StatusCheckRunnable和Dispatcher（WorkerDetector放在《较真儿学源码系列-PowerJob MapReduce源码分析》中分析）：
       

      2.3.1.1 StatusCheckRunnable
       

      1. /**
      2.  * StatusCheckRunnable:
      3.  */
      4. @Override
      5. public void run() {
      6.     try {
      7.         innerRun();
      8.     } catch (Exception e) {
      9.         log.warn("[TaskTracker-{}] status checker execute failed, please fix the bug (@tjq)!", instanceId, e);
      10.     }
      11. }
      12.  
      13. /**
      14.  * 第7行代码处：
      15.  */
      16. @SuppressWarnings("squid:S3776")
      17. private void innerRun() {
      18.  
      19.     //获取任务实例产生的各个Task状态，用于分析任务实例执行情况
      20.     InstanceStatisticsHolder holder = getInstanceStatisticsHolder(instanceId);
      21.  
      22.     long finishedNum = holder.succeedNum + holder.failedNum;
      23.     long unfinishedNum = holder.waitingDispatchNum + holder.workerUnreceivedNum + holder.receivedNum + holder.runningNum;
      24.  
      25.     log.debug("[TaskTracker-{}] status check result: {}", instanceId, holder);
      26.  
      27.     //组装上报参数
      28.     TaskTrackerReportInstanceStatusReq req = new TaskTrackerReportInstanceStatusReq();
      29.     req.setAppId(workerRuntime.getAppId());
      30.     req.setJobId(instanceInfo.getJobId());
      31.     req.setInstanceId(instanceId);
      32.     req.setWfInstanceId(instanceInfo.getWfInstanceId());
      33.     req.setTotalTaskNum(finishedNum + unfinishedNum);
      34.     req.setSucceedTaskNum(holder.succeedNum);
      35.     req.setFailedTaskNum(holder.failedNum);
      36.     req.setReportTime(System.currentTimeMillis());
      37.     req.setStartTime(createTime);
      38.     req.setSourceAddress(workerRuntime.getWorkerAddress());
      39.  
      40.     boolean success = false;
      41.     String result = null;
      42.  
      43.     // 2. 如果未完成任务数为0，判断是否真正结束，并获取真正结束任务的执行结果
      44.     if (unfinishedNum == 0) {
      45.  
      46.         // 数据库中一个任务都没有，说明根任务创建失败，该任务实例失败
      47.         if (finishedNum == 0) {
      48.             finished.set(true);
      49.             result = SystemInstanceResult.TASK_INIT_FAILED;
      50.         } else {
      51.             ExecuteType executeType = ExecuteType.valueOf(instanceInfo.getExecuteType());
      52.  
      53.             switch (executeType) {
      54.  
      55.                 // STANDALONE 只有一个任务，完成即结束
      56.                 case STANDALONE:
      57.                     finished.set(true);
      58.                     List allTask = taskPersistenceService.getAllTask(instanceId, instanceId);
      59.                     if (CollectionUtils.isEmpty(allTask) || allTask.size() > 1) {
      60.                         result = SystemInstanceResult.UNKNOWN_BUG;
      61.                         log.warn("[TaskTracker-{}] there must have some bug in TaskTracker.", instanceId);
      62.                     } else {
      63.                         result = allTask.get(0).getResult();
      64.                         success = allTask.get(0).getStatus() == TaskStatus.WORKER_PROCESS_SUCCESS.getValue();
      65.                     }
      66.                     break;
      67.                 case MAP:
      68.                     //...
      69.                     break;
      70.                 default:
      71.                     //...
      72.             }
      73.         }
      74.     }
      75.  
      76.     // 3. 检查任务实例整体是否超时
      77.     if (isTimeout()) {
      78.         finished.set(true);
      79.         success = false;
      80.         result = SystemInstanceResult.INSTANCE_EXECUTE_TIMEOUT;
      81.     }
      82.  
      83.     // 4. 执行完毕，报告服务器
      84.     if (finished.get()) {
      85.         req.setResult(result);
      86.         // 上报追加的工作流上下文信息
      87.         req.setAppendedWfContext(appendedWfContext);
      88.         req.setInstanceStatus(success ? InstanceStatus.SUCCEED.getV() : InstanceStatus.FAILED.getV());
      89.         reportFinalStatusThenDestroy(workerRuntime, req);
      90.         return;
      91.     }
      92.  
      93.     // 5. 未完成，上报状态
      94.     req.setInstanceStatus(InstanceStatus.RUNNING.getV());
      95.     TransportUtils.ttReportInstanceStatus(req, workerRuntime.getServerDiscoveryService().getCurrentServerAddress(), workerRuntime.getTransporter());
      96.  
      97.     // 6.1 定期检查 -> 重试派发后未确认的任务
      98.     long currentMS = System.currentTimeMillis();
      99.     if (holder.workerUnreceivedNum != 0) {
      100.         taskPersistenceService.getTaskByStatus(instanceId, TaskStatus.DISPATCH_SUCCESS_WORKER_UNCHECK, 100).forEach(uncheckTask -> {
      101.  
      102.             long elapsedTime = currentMS - uncheckTask.getLastModifiedTime();
      103.             if (elapsedTime > DISPATCH_TIME_OUT_MS) {
      104.  
      105.                 TaskDO updateEntity = new TaskDO();
      106.                 updateEntity.setStatus(TaskStatus.WAITING_DISPATCH.getValue());
      107.                 // 特殊任务只能本机执行
      108.                 if (!TaskConstant.LAST_TASK_NAME.equals(uncheckTask.getTaskName())) {
      109.                     updateEntity.setAddress(RemoteConstant.EMPTY_ADDRESS);
      110.                 }
      111.                 // 失败次数 + 1
      112.                 updateEntity.setFailedCnt(uncheckTask.getFailedCnt() + 1);
      113.  
      114.                 taskPersistenceService.updateTask(instanceId, uncheckTask.getTaskId(), updateEntity);
      115.  
      116.                 log.warn("[TaskTracker-{}] task(id={},name={}) try to dispatch again due to unreceived the response from ProcessorTracker.",
      117.                         instanceId, uncheckTask.getTaskId(), uncheckTask.getTaskName());
      118.             }
      119.  
      120.         });
      121.     }
      122.  
      123.     // 6.2 定期检查 -> 重新执行被派发到宕机ProcessorTracker上的任务
      124.     List disconnectedPTs = ptStatusHolder.getAllDisconnectedProcessorTrackers();
      125.     if (!disconnectedPTs.isEmpty()) {
      126.         log.warn("[TaskTracker-{}] some ProcessorTracker disconnected from TaskTracker,their address is {}.", instanceId, disconnectedPTs);
      127.         if (taskPersistenceService.updateLostTasks(instanceId, disconnectedPTs, true)) {
      128.             ptStatusHolder.remove(disconnectedPTs);
      129.             log.warn("[TaskTracker-{}] removed these ProcessorTracker from StatusHolder: {}", instanceId, disconnectedPTs);
      130.         }
      131.     }
      132. }
       

              上报到服务端的逻辑这里就不再看了（更新instance_info实例表数据）。整体上来说，StatusCheckRunnable的作用就是检查任务的执行情况，并上报到服务端。
       

      2.3.1.2 Dispatcher
       

      1. /**
      2.  * Dispatcher:
      3.  */
      4. @Override
      5. public void run() {
      6.  
      7.     if (finished.get()) {
      8.         return;
      9.     }
      10.  
      11.     Stopwatch stopwatch = Stopwatch.createStarted();
      12.  
      13.     // 1. 获取可以派发任务的 ProcessorTracker
      14.     List availablePtIps = ptStatusHolder.getAvailableProcessorTrackers();
      15.  
      16.     // 2. 没有可用 ProcessorTracker，本次不派发
      17.     if (availablePtIps.isEmpty()) {
      18.         log.debug("[TaskTracker-{}] no available ProcessorTracker now.", instanceId);
      19.         return;
      20.     }
      21.  
      22.     // 3. 避免大查询，分批派发任务
      23.     long currentDispatchNum = 0;
      24.     //这里需要留意下，最大分发次数=可用ProcessorTracker数量*实例并发度*2
      25.     long maxDispatchNum = availablePtIps.size() * instanceInfo.getThreadConcurrency() * 2L;
      26.     AtomicInteger index = new AtomicInteger(0);
      27.  
      28.     // 4. 循环查询数据库，获取需要派发的任务
      29.     while (maxDispatchNum > currentDispatchNum) {
      30.  
      31.         //每次最多查询100个任务
      32.         int dbQueryLimit = Math.min(DB_QUERY_LIMIT, (int) maxDispatchNum);
      33.         //获取等待调度器调度的任务
      34.         List needDispatchTasks = taskPersistenceService.getTaskByStatus(instanceId, TaskStatus.WAITING_DISPATCH, dbQueryLimit);
      35.         currentDispatchNum += needDispatchTasks.size();
      36.  
      37.         needDispatchTasks.forEach(task -> {
      38.             // 获取 ProcessorTracker 地址，如果 Task 中自带了 Address，则使用该 Address
      39.             String ptAddress = task.getAddress();
      40.             if (StringUtils.isEmpty(ptAddress) || RemoteConstant.EMPTY_ADDRESS.equals(ptAddress)) {
      41.                 //否则，从可用ProcessorTracker中取余获取一个
      42.                 ptAddress = availablePtIps.get(index.getAndIncrement() % availablePtIps.size());
      43.             }
      44.             //分发任务
      45.             dispatchTask(task, ptAddress);
      46.         });
      47.  
      48.         // 数量不足 或 查询失败，则终止循环
      49.         if (needDispatchTasks.size() < dbQueryLimit) {
      50.             break;
      51.         }
      52.     }
      53.  
      54.     log.debug("[TaskTracker-{}] dispatched {} tasks,using time {}.", instanceId, currentDispatchNum, stopwatch.stop());
      55. }
      56.  
      57. /**
      58.  * ProcessorTrackerStatusHolder:
      59.  * 第14行代码处：
      60.  * 获取可用 ProcessorTracker 的IP地址
      61.  */
      62. public List getAvailableProcessorTrackers() {
      63.  
      64.     List result = Lists.newLinkedList();
      65.     address2Status.forEach((address, ptStatus) -> {
      66.         if (ptStatus.available()) {
      67.             result.add(address);
      68.         }
      69.     });
      70.     return result;
      71. }
      72.  
      73. /**
      74.  * ProcessorTrackerStatus:
      75.  * 第66行代码处：
      76.  * 是否可用
      77.  */
      78. public boolean available() {
      79.  
      80.     // 未曾派发过，默认可用
      81.     if (!dispatched) {
      82.         return true;
      83.     }
      84.  
      85.     // 已派发但未收到响应，则不可用
      86.     if (!connected) {
      87.         return false;
      88.     }
      89.  
      90.     // 长时间未收到心跳消息，则不可用
      91.     if (isTimeout()) {
      92.         return false;
      93.     }
      94.  
      95.     // 留有过多待处理任务，则不可用
      96.     if (remainTaskNum >= DISPATCH_THRESHOLD) {
      97.         return false;
      98.     }
      99.  
      100.     // TODO：后续考虑加上机器健康度等信息
      101.  
      102.     return true;
      103. }
      104.  
      105. /**
      106.  * 第91行代码处：
      107.  * 是否超时（超过一定时间没有收到心跳）
      108.  */
      109. public boolean isTimeout() {
      110.     if (dispatched) {
      111.         //系统当前时间-上次活跃时间>心跳超时时间
      112.         return System.currentTimeMillis() - lastActiveTime > HEARTBEAT_TIMEOUT_MS;
      113.     }
      114.     // 未曾派发过任务的机器，不用处理
      115.     return false;
      116. }
      117.  
      118. /**
      119.  * HeavyTaskTracker:
      120.  * 第45行代码处：
      121.  * 派发任务到 ProcessorTracker
      122.  *
      123.  * @param task                    需要被执行的任务
      124.  * @param processorTrackerAddress ProcessorTracker的地址（IP:Port）
      125.  */
      126. protected void dispatchTask(TaskDO task, String processorTrackerAddress) {
      127.  
      128.     // 1. 持久化，更新数据库（如果更新数据库失败，可能导致重复执行，先不处理）
      129.     TaskDO updateEntity = new TaskDO();
      130.     updateEntity.setStatus(TaskStatus.DISPATCH_SUCCESS_WORKER_UNCHECK.getValue());
      131.     // 写入处理该任务的 ProcessorTracker
      132.     updateEntity.setAddress(processorTrackerAddress);
      133.     boolean success = taskPersistenceService.updateTask(instanceId, task.getTaskId(), updateEntity);
      134.     if (!success) {
      135.         log.warn("[TaskTracker-{}] dispatch task(taskId={},taskName={}) failed due to update task status failed.", instanceId, task.getTaskId(), task.getTaskName());
      136.         return;
      137.     }
      138.  
      139.     // 2. 更新 ProcessorTrackerStatus 状态
      140.     ptStatusHolder.getProcessorTrackerStatus(processorTrackerAddress).setDispatched(true);
      141.     // 3. 初始化缓存
      142.     taskId2BriefInfo.put(task.getTaskId(), new TaskBriefInfo(task.getTaskId(), TaskStatus.DISPATCH_SUCCESS_WORKER_UNCHECK, -1L));
      143.  
      144.     // 4. 任务派发
      145.     TaskTrackerStartTaskReq startTaskReq = new TaskTrackerStartTaskReq(instanceInfo, task, workerRuntime.getWorkerAddress());
      146.     TransportUtils.ttStartPtTask(startTaskReq, processorTrackerAddress, workerRuntime.getTransporter());
      147.  
      148.     log.debug("[TaskTracker-{}] dispatch task(taskId={},taskName={}) successfully.", instanceId, task.getTaskId(), task.getTaskName());
      149. }
      150.  
      151. /**
      152.  * TransportUtils:
      153.  * 第146行代码处：
      154.  */
      155. public static void ttStartPtTask(TaskTrackerStartTaskReq req, String address, Transporter transporter) {
      156.     final URL url = easyBuildUrl(ServerType.WORKER, WPT_PATH, WPT_HANDLER_START_TASK, address);
      157.     transporter.tell(url, req);
      158. }
      159.  
      160. /**
      161.  * ProcessorTrackerActor:
      162.  * 第157行代码处：
      163.  * 处理来自TaskTracker的task执行请求
      164.  *
      165.  * @param req 请求
      166.  */
      167. @Handler(path = RemoteConstant.WPT_HANDLER_START_TASK, processType = ProcessType.NO_BLOCKING)
      168. public void onReceiveTaskTrackerStartTaskReq(TaskTrackerStartTaskReq req) {
      169.  
      170.     Long instanceId = req.getInstanceInfo().getInstanceId();
      171.  
      172.     // 创建 ProcessorTracker 一定能成功
      173.     ProcessorTracker processorTracker = ProcessorTrackerManager.getProcessorTracker(
      174.             instanceId,
      175.             req.getTaskTrackerAddress(),
      176.             () -> new ProcessorTracker(req, workerRuntime));
      177.  
      178.     TaskDO task = new TaskDO();
      179.  
      180.     task.setTaskId(req.getTaskId());
      181.     task.setTaskName(req.getTaskName());
      182.     task.setTaskContent(req.getTaskContent());
      183.     task.setFailedCnt(req.getTaskCurrentRetryNums());
      184.     task.setSubInstanceId(req.getSubInstanceId());
      185.  
      186.     processorTracker.submitTask(task);
      187. }
      188.  
      189. /**
      190.  * ProcessorTracker:
      191.  * 提交任务到线程池执行
      192.  * 1.0版本：TaskTracker有任务就dispatch，导致 ProcessorTracker 本地可能堆积过多的任务，造成内存压力。为此 ProcessorTracker 在线程
      193.  * 池队列堆积到一定程度时，会将数据持久化到DB，然后通过异步线程定时从数据库中取回任务，重新提交执行。
      194.  * 联动：数据库的SPID设计、TaskStatus段落设计等，全部取消...
      195.  * last commitId: 341953aceceafec0fbe7c3d9a3e26451656b945e
      196.  * 2.0版本：ProcessorTracker定时向TaskTracker发送心跳消息，心跳消息中包含了当前线程池队列任务个数，TaskTracker根据ProcessorTracker
      197.  * 的状态判断能否继续派发任务。因此，ProcessorTracker本地不会堆积过多任务，故删除 持久化机制 ╥﹏╥...！
      198.  *
      199.  * @param newTask 需要提交到线程池执行的任务
      200.  */
      201. public void submitTask(TaskDO newTask) {
      202.  
      203.     // 一旦 ProcessorTracker 出现异常，所有提交到此处的任务直接返回失败，防止形成死锁
      204.     // 死锁分析：TT创建PT，PT创建失败，无法定期汇报心跳，TT长时间未收到PT心跳，认为PT宕机（确实宕机了），无法选择可用的PT再次派发任务，死锁形成，GG斯密达 T_T
      205.     if (lethal) {
      206.         ProcessorReportTaskStatusReq report = new ProcessorReportTaskStatusReq()
      207.                 .setInstanceId(instanceId)
      208.                 .setSubInstanceId(newTask.getSubInstanceId())
      209.                 .setTaskId(newTask.getTaskId())
      210.                 .setStatus(TaskStatus.WORKER_PROCESS_FAILED.getValue())
      211.                 .setResult(lethalReason)
      212.                 .setReportTime(System.currentTimeMillis());
      213.  
      214.         TransportUtils.ptReportTask(report, taskTrackerAddress, workerRuntime);
      215.         return;
      216.     }
      217.  
      218.     boolean success = false;
      219.     // 1. 设置值并提交执行
      220.     newTask.setInstanceId(instanceInfo.getInstanceId());
      221.     newTask.setAddress(taskTrackerAddress);
      222.  
      223.     HeavyProcessorRunnable heavyProcessorRunnable = new HeavyProcessorRunnable(instanceInfo, taskTrackerAddress, newTask, processorBean, omsLogger, statusReportRetryQueue, workerRuntime);
      224.     try {
      225.         threadPool.submit(heavyProcessorRunnable);
      226.         success = true;
      227.     } catch (RejectedExecutionException ignore) {
      228.         log.warn("[ProcessorTracker-{}] submit task(taskId={},taskName={}) to ThreadPool failed due to ThreadPool has too much task waiting to process, this task will dispatch to other ProcessorTracker.",
      229.                 instanceId, newTask.getTaskId(), newTask.getTaskName());
      230.     } catch (Exception e) {
      231.         log.error("[ProcessorTracker-{}] submit task(taskId={},taskName={}) to ThreadPool failed.", instanceId, newTask.getTaskId(), newTask.getTaskName(), e);
      232.     }
      233.  
      234.     // 2. 回复接收成功
      235.     if (success) {
      236.         ProcessorReportTaskStatusReq reportReq = new ProcessorReportTaskStatusReq();
      237.         reportReq.setInstanceId(instanceId);
      238.         reportReq.setSubInstanceId(newTask.getSubInstanceId());
      239.         reportReq.setTaskId(newTask.getTaskId());
      240.         reportReq.setStatus(TaskStatus.WORKER_RECEIVED.getValue());
      241.         reportReq.setReportTime(System.currentTimeMillis());
      242.  
      243.         TransportUtils.ptReportTask(reportReq, taskTrackerAddress, workerRuntime);
      244.  
      245.         log.debug("[ProcessorTracker-{}] submit task(taskId={}, taskName={}) success, current queue size: {}.",
      246.                 instanceId, newTask.getTaskId(), newTask.getTaskName(), threadPool.getQueue().size());
      247.     }
      248. }
       

              可以看到，在最后，Dispatcher会创建出一个HeavyProcessorRunnable的线程来执行，里面存放着需要执行的任务实例、执行地址、执行处理器等信息。接下来看下其实现：
       

      1. /**
      2.  * HeavyProcessorRunnable:
      3.  */
      4. @Override
      5. @SuppressWarnings("squid:S2142")
      6. public void run() {
      7.     // 切换线程上下文类加载器（否则用的是 Worker 类加载器，不存在容器类，在序列化/反序列化时会报 ClassNotFoundException）
      8.     Thread.currentThread().setContextClassLoader(processorBean.getClassLoader());
      9.     try {
      10.         innerRun();
      11.     } catch (InterruptedException ignore) {
      12.         // ignore
      13.     } catch (Throwable e) {
      14.         reportStatus(TaskStatus.WORKER_PROCESS_FAILED, e.toString(), null, null);
      15.         log.error("[ProcessorRunnable-{}] execute failed, please contact the author(@KFCFans) to fix the bug!", task.getInstanceId(), e);
      16.     } finally {
      17.         ThreadLocalStore.clear();
      18.     }
      19. }
      20.  
      21. /**
      22.  * 第10行代码处：
      23.  */
      24. public void innerRun() throws InterruptedException {
      25.  
      26.     //获取执行处理器
      27.     final BasicProcessor processor = processorBean.getProcessor();
      28.  
      29.     String taskId = task.getTaskId();
      30.     Long instanceId = task.getInstanceId();
      31.  
      32.     log.debug("[ProcessorRunnable-{}] start to run task(taskId={}&taskName={})", instanceId, taskId, task.getTaskName());
      33.     //缓存
      34.     ThreadLocalStore.setTask(task);
      35.     ThreadLocalStore.setRuntimeMeta(workerRuntime);
      36.  
      37.     // 0. 构造任务上下文
      38.     WorkflowContext workflowContext = constructWorkflowContext();
      39.     TaskContext taskContext = constructTaskContext();
      40.     taskContext.setWorkflowContext(workflowContext);
      41.     // 1. 上报执行信息
      42.     reportStatus(TaskStatus.WORKER_PROCESSING, null, null, null);
      43.  
      44.     ProcessResult processResult;
      45.     ExecuteType executeType = ExecuteType.valueOf(instanceInfo.getExecuteType());
      46.  
      47.     // 2. 根任务 & 广播执行 特殊处理
      48.     if (TaskConstant.ROOT_TASK_NAME.equals(task.getTaskName()) && executeType == ExecuteType.BROADCAST) {
      49.         // 广播执行：先选本机执行 preProcess，完成后 TaskTracker 再为所有 Worker 生成子 Task
      50.         handleBroadcastRootTask(instanceId, taskContext);
      51.         return;
      52.     }
      53.  
      54.     // 3. 最终任务特殊处理（一定和 TaskTracker 处于相同的机器）
      55.     if (TaskConstant.LAST_TASK_NAME.equals(task.getTaskName())) {
      56.         handleLastTask(taskId, instanceId, taskContext, executeType);
      57.         return;
      58.     }
      59.  
      60.     // 4. 正式提交运行
      61.     try {
      62.         //这里的process方法也就是我们自己写的业务方法
      63.         processResult = processor.process(taskContext);
      64.         if (processResult == null) {
      65.             processResult = new ProcessResult(false, "ProcessResult can't be null");
      66.         }
      67.     } catch (Throwable e) {
      68.         log.warn("[ProcessorRunnable-{}] task(id={},name={}) process failed.", instanceId, taskContext.getTaskId(), taskContext.getTaskName(), e);
      69.         processResult = new ProcessResult(false, e.toString());
      70.     }
      71.     //上报执行结果
      72.     reportStatus(processResult.isSuccess() ? TaskStatus.WORKER_PROCESS_SUCCESS : TaskStatus.WORKER_PROCESS_FAILED, suit(processResult.getMsg()), null, workflowContext.getAppendedContextData());
      73. }
       

              Dispatcher的作用就是分发任务，并执行任务。也就是会有一个单独的线程（HeavyProcessorRunnable），会定时从任务表中拿取任务，然后执行我们自己实现的process方法，并会在执行前和执行后上报执行信息。
       

      2.3.2 CheckRunningInstance
       

      1. /**
      2.  * InstanceStatusCheckService:
      3.  * 检查运行中的实例
      4.  * RUNNING 超时：TaskTracker down，断开与 server 的心跳连接
      5.  */
      6. public void checkRunningInstance() {
      7.     Stopwatch stopwatch = Stopwatch.createStarted();
      8.     // 查询 DB 获取该 Server 需要负责的 AppGroup
      9.     //获取appId
      10.     List allAppIds = appInfoRepository.listAppIdByCurrentServer(transportService.defaultProtocol().getAddress());
      11.     if (CollectionUtils.isEmpty(allAppIds)) {
      12.         log.info("[InstanceStatusChecker] current server has no app's job to check");
      13.         return;
      14.     }
      15.     try {
      16.         // 检查 RUNNING 状态的任务（一定时间没收到 TaskTracker 的状态报告，视为失败）
      17.         Lists.partition(allAppIds, MAX_BATCH_NUM_APP).forEach(this::handleRunningInstance);
      18.     } catch (Exception e) {
      19.         log.error("[InstanceStatusChecker] RunningInstance status check failed.", e);
      20.     }
      21.     log.info("[InstanceStatusChecker] RunningInstance status check using {}.", stopwatch.stop());
      22. }
      23.  
      24. /**
      25.  * 第17行代码处：
      26.  */
      27. private void handleRunningInstance(List partAppIds) {
      28.     // 3. 检查 RUNNING 状态的任务（一定时间没收到 TaskTracker 的状态报告，视为失败）
      29.     long threshold = System.currentTimeMillis() - RUNNING_TIMEOUT_MS;
      30.     //查找修改时间距离现在超过1分钟的实例
      31.     List failedInstances = instanceInfoRepository.selectBriefInfoByAppIdInAndStatusAndGmtModifiedBefore(partAppIds, InstanceStatus.RUNNING.getV(), new Date(threshold), PageRequest.of(0, MAX_BATCH_NUM_INSTANCE));
      32.     while (!failedInstances.isEmpty()) {
      33.         // collect job id
      34.         Set jobIds = failedInstances.stream().map(BriefInstanceInfo::getJobId).collect(Collectors.toSet());
      35.         // query job info and map
      36.         Map jobInfoMap = jobInfoRepository.findByIdIn(jobIds).stream().collect(Collectors.toMap(JobInfoDO::getId, e -> e));
      37.         log.warn("[InstanceStatusCheckService] find some instances have not received status report for a long time : {}", failedInstances.stream().map(BriefInstanceInfo::getInstanceId).collect(Collectors.toList()));
      38.         failedInstances.forEach(instance -> {
      39.             Optional jobInfoOpt = Optional.ofNullable(jobInfoMap.get(instance.getJobId()));
      40.             if (!jobInfoOpt.isPresent()) {
      41.                 final Optional opt = instanceInfoRepository.findById(instance.getId());
      42.                 opt.ifPresent(e -> updateFailedInstance(e, SystemInstanceResult.REPORT_TIMEOUT));
      43.                 return;
      44.             }
      45.             TimeExpressionType timeExpressionType = TimeExpressionType.of(jobInfoOpt.get().getTimeExpressionType());
      46.             SwitchableStatus switchableStatus = SwitchableStatus.of(jobInfoOpt.get().getStatus());
      47.             // 如果任务已关闭，则不进行重试，将任务置为失败即可；秒级任务也直接置为失败，由派发器重新调度
      48.             if (switchableStatus != SwitchableStatus.ENABLE || TimeExpressionType.FREQUENT_TYPES.contains(timeExpressionType.getV())) {
      49.                 final Optional opt = instanceInfoRepository.findById(instance.getId());
      50.                 opt.ifPresent(e -> updateFailedInstance(e, SystemInstanceResult.REPORT_TIMEOUT));
      51.                 return;
      52.             }
      53.             // CRON 和 API一样，失败次数 + 1，根据重试配置进行重试
      54.             if (instance.getRunningTimes() < jobInfoOpt.get().getInstanceRetryNum()) {
      55.                 //更新实例表状态为等待派发
      56.                 dispatchService.redispatchAsync(instance.getInstanceId(), InstanceStatus.RUNNING.getV());
      57.             } else {
      58.                 final Optional opt = instanceInfoRepository.findById(instance.getId());
      59.                 opt.ifPresent(e -> updateFailedInstance(e, SystemInstanceResult.REPORT_TIMEOUT));
      60.             }
      61.         });
      62.         threshold = System.currentTimeMillis() - RUNNING_TIMEOUT_MS;
      63.         //重新查找修改时间距离现在超过1分钟的实例，并循环
      64.         failedInstances = instanceInfoRepository.selectBriefInfoByAppIdInAndStatusAndGmtModifiedBefore(partAppIds, InstanceStatus.RUNNING.getV(), new Date(threshold), PageRequest.of(0, MAX_BATCH_NUM_INSTANCE));
      65.     }
      66.  
      67. }
       

       

      3 客户端
       

      3.1 PowerJobSpringWorker
       

              客户端的启动类同样没有什么逻辑，查看下初始化类PowerJobSpringWorker：
       

      1. /**
      2.  * PowerJobSpringWorker:
      3.  */
      4. @Override
      5. public void afterPropertiesSet() throws Exception {
      6.     powerJobWorker = new PowerJobWorker(config);
      7.     powerJobWorker.init();
      8. }
      9.  
      10. /**
      11.  * PowerJobWorker:
      12.  */
      13. public void init() throws Exception {
      14.  
      15.     if (!initialized.compareAndSet(false, true)) {
      16.         log.warn("[PowerJobWorker] please do not repeat the initialization");
      17.         return;
      18.     }
      19.  
      20.     Stopwatch stopwatch = Stopwatch.createStarted();
      21.     log.info("[PowerJobWorker] start to initialize PowerJobWorker...");
      22.  
      23.     PowerJobWorkerConfig config = workerRuntime.getWorkerConfig();
      24.     CommonUtils.requireNonNull(config, "can't find PowerJobWorkerConfig, please set PowerJobWorkerConfig first");
      25.  
      26.     try {
      27.         //打印banner
      28.         PowerBannerPrinter.print();
      29.         // 校验 appName
      30.         if (!config.isEnableTestMode()) {
      31.             assertAppName();
      32.         } else {
      33.             log.warn("[PowerJobWorker] using TestMode now, it's dangerous if this is production env.");
      34.         }
      35.  
      36.         // 初始化元数据
      37.         String workerAddress = NetUtils.getLocalHost() + ":" + config.getPort();
      38.         workerRuntime.setWorkerAddress(workerAddress);
      39.  
      40.         // 初始化 线程池
      41.         final ExecutorManager executorManager = new ExecutorManager(workerRuntime.getWorkerConfig());
      42.         workerRuntime.setExecutorManager(executorManager);
      43.  
      44.         // 初始化 ProcessorLoader
      45.         //动态加载类的工厂，可以不依赖于Spring的实现
      46.         ProcessorLoader processorLoader = buildProcessorLoader(workerRuntime);
      47.         workerRuntime.setProcessorLoader(processorLoader);
      48.  
      49.         // 初始化 actor
      50.         TaskTrackerActor taskTrackerActor = new TaskTrackerActor(workerRuntime);
      51.         ProcessorTrackerActor processorTrackerActor = new ProcessorTrackerActor(workerRuntime);
      52.         WorkerActor workerActor = new WorkerActor(workerRuntime, taskTrackerActor);
      53.  
      54.         // 初始化通讯引擎
      55.         EngineConfig engineConfig = new EngineConfig()
      56.                 .setType(config.getProtocol().name())
      57.                 .setServerType(ServerType.WORKER)
      58.                 .setBindAddress(new Address().setHost(NetUtils.getLocalHost()).setPort(config.getPort()))
      59.                 .setActorList(Lists.newArrayList(taskTrackerActor, processorTrackerActor, workerActor));
      60.  
      61.         //之前服务端启动的时候也会执行PowerJobRemoteEngine.start方法
      62.         EngineOutput engineOutput = remoteEngine.start(engineConfig);
      63.         workerRuntime.setTransporter(engineOutput.getTransporter());
      64.  
      65.         // 连接 server
      66.         ServerDiscoveryService serverDiscoveryService = new ServerDiscoveryService(workerRuntime.getAppId(), workerRuntime.getWorkerConfig());
      67.  
      68.         serverDiscoveryService.start(workerRuntime.getExecutorManager().getCoreExecutor());
      69.         workerRuntime.setServerDiscoveryService(serverDiscoveryService);
      70.  
      71.         log.info("[PowerJobWorker] PowerJobRemoteEngine initialized successfully.");
      72.  
      73.         // 初始化日志系统
      74.         OmsLogHandler omsLogHandler = new OmsLogHandler(workerAddress, workerRuntime.getTransporter(), serverDiscoveryService);
      75.         workerRuntime.setOmsLogHandler(omsLogHandler);
      76.  
      77.         // 初始化存储
      78.         TaskPersistenceService taskPersistenceService = new TaskPersistenceService(workerRuntime.getWorkerConfig().getStoreStrategy());
      79.         taskPersistenceService.init();
      80.         workerRuntime.setTaskPersistenceService(taskPersistenceService);
      81.         log.info("[PowerJobWorker] local storage initialized successfully.");
      82.  
      83.  
      84.         // 初始化定时任务
      85.         workerRuntime.getExecutorManager().getCoreExecutor().scheduleAtFixedRate(new WorkerHealthReporter(workerRuntime), 0, config.getHealthReportInterval(), TimeUnit.SECONDS);
      86.         workerRuntime.getExecutorManager().getCoreExecutor().scheduleWithFixedDelay(omsLogHandler.logSubmitter, 0, 5, TimeUnit.SECONDS);
      87.  
      88.         log.info("[PowerJobWorker] PowerJobWorker initialized successfully, using time: {}, congratulations!", stopwatch);
      89.     } catch (Exception e) {
      90.         log.error("[PowerJobWorker] initialize PowerJobWorker failed, using {}.", stopwatch, e);
      91.         throw e;
      92.     }
      93. }
      94.  
      95. /**
      96.  * 第31行代码处：
      97.  */
      98. @SuppressWarnings("rawtypes")
      99. private void assertAppName() {
      100.  
      101.     PowerJobWorkerConfig config = workerRuntime.getWorkerConfig();
      102.     String appName = config.getAppName();
      103.     Objects.requireNonNull(appName, "appName can't be empty!");
      104.  
      105.     String url = "http://%s/server/assert?appName=%s";
      106.     for (String server : config.getServerAddress()) {
      107.         String realUrl = String.format(url, server, appName);
      108.         try {
      109.             //客户端启动的时候连接一下服务端，看是否能连通
      110.             String resultDTOStr = CommonUtils.executeWithRetry0(() -> HttpUtils.get(realUrl));
      111.             ResultDTO resultDTO = JsonUtils.parseObject(resultDTOStr, ResultDTO.class);
      112.             if (resultDTO.isSuccess()) {
      113.                 Long appId = Long.valueOf(resultDTO.getData().toString());
      114.                 log.info("[PowerJobWorker] assert appName({}) succeed, the appId for this application is {}.", appName, appId);
      115.                 workerRuntime.setAppId(appId);
      116.                 return;
      117.             } else {
      118.                 log.error("[PowerJobWorker] assert appName failed, this appName is invalid, please register the appName {} first.", appName);
      119.                 throw new PowerJobException(resultDTO.getMessage());
      120.             }
      121.         } catch (PowerJobException oe) {
      122.             throw oe;
      123.         } catch (Exception ignore) {
      124.            log.warn("[PowerJobWorker] assert appName by url({}) failed, please check the server address.", realUrl);
      125.         }
      126.     }
      127.     log.error("[PowerJobWorker] no available server in {}.", config.getServerAddress());
      128.     throw new PowerJobException("no server available!");
      129. }
      130.  
      131. /**
      132.  * ExecutorManager:
      133.  * 第41行代码处：
      134.  */
      135. public ExecutorManager(PowerJobWorkerConfig workerConfig) {
      136.  
      137.     //可用cpu数
      138.     final int availableProcessors = Runtime.getRuntime().availableProcessors();
      139.     // 初始化定时线程池
      140.     ThreadFactory coreThreadFactory = new ThreadFactoryBuilder().setNameFormat("powerjob-worker-core-%d").build();
      141.     coreExecutor = new ScheduledThreadPoolExecutor(3, coreThreadFactory);
      142.  
      143.     ThreadFactory lightTaskReportFactory = new ThreadFactoryBuilder().setNameFormat("powerjob-worker-light-task-status-check-%d").build();
      144.     // 都是 io 密集型任务
      145.     lightweightTaskStatusCheckExecutor = new ScheduledThreadPoolExecutor(availableProcessors * 10, lightTaskReportFactory);
      146.  
      147.     ThreadFactory lightTaskExecuteFactory = new ThreadFactoryBuilder().setNameFormat("powerjob-worker-light-task-execute-%d").build();
      148.     // 大部分任务都是 io 密集型
      149.     lightweightTaskExecutorService = new ThreadPoolExecutor(availableProcessors * 10, availableProcessors * 10, 120L, TimeUnit.SECONDS,
      150.             new ArrayBlockingQueue<>((workerConfig.getMaxLightweightTaskNum() * 2), true), lightTaskExecuteFactory, new ThreadPoolExecutor.AbortPolicy());
      151.  
      152. }
      153.  
      154. /**
      155.  * ServerDiscoveryService:
      156.  * 第68行代码处：
      157.  */
      158. public void start(ScheduledExecutorService timingPool) {
      159.     this.currentServerAddress = discovery();
      160.     if (StringUtils.isEmpty(this.currentServerAddress) && !config.isEnableTestMode()) {
      161.         throw new PowerJobException("can't find any available server, this worker has been quarantined.");
      162.     }
      163.     // 这里必须保证成功
      164.     timingPool.scheduleAtFixedRate(() -> {
      165.                 try {
      166.                     this.currentServerAddress = discovery();
      167.                 } catch (Exception e) {
      168.                     log.error("[PowerDiscovery] fail to discovery server!", e);
      169.                 }
      170.             }
      171.             , 10, 10, TimeUnit.SECONDS);
      172. }
      173.  
      174. /**
      175.  * 第159行代码处和第166行代码处：
      176.  */
      177. private String discovery() {
      178.  
      179.     if (ip2Address.isEmpty()) {
      180.         config.getServerAddress().forEach(x -> ip2Address.put(x.split(":")[0], x));
      181.     }
      182.  
      183.     String result = null;
      184.  
      185.     // 先对当前机器发起请求
      186.     String currentServer = currentServerAddress;
      187.     if (!StringUtils.isEmpty(currentServer)) {
      188.         String ip = currentServer.split(":")[0];
      189.         // 直接请求当前Server的HTTP服务，可以少一次网络开销，减轻Server负担
      190.         String firstServerAddress = ip2Address.get(ip);
      191.         if (firstServerAddress != null) {
      192.             result = acquire(firstServerAddress);
      193.         }
      194.     }
      195.  
      196.     for (String httpServerAddress : config.getServerAddress()) {
      197.         if (StringUtils.isEmpty(result)) {
      198.             result = acquire(httpServerAddress);
      199.         } else {
      200.             break;
      201.         }
      202.     }
      203.  
      204.     if (StringUtils.isEmpty(result)) {
      205.         log.warn("[PowerDiscovery] can't find any available server, this worker has been quarantined.");
      206.  
      207.         // 在 Server 高可用的前提下，连续失败多次，说明该节点与外界失联，Server已经将秒级任务转移到其他Worker，需要杀死本地的任务
      208.         if (FAILED_COUNT++ > MAX_FAILED_COUNT) {
      209.  
      210.             log.warn("[PowerDiscovery] can't find any available server for 3 consecutive times, It's time to kill all frequent job in this worker.");
      211.             List frequentInstanceIds = HeavyTaskTrackerManager.getAllFrequentTaskTrackerKeys();
      212.             if (!CollectionUtils.isEmpty(frequentInstanceIds)) {
      213.                 frequentInstanceIds.forEach(instanceId -> {
      214.                     HeavyTaskTracker taskTracker = HeavyTaskTrackerManager.removeTaskTracker(instanceId);
      215.                     taskTracker.destroy();
      216.                     log.warn("[PowerDiscovery] kill frequent instance(instanceId={}) due to can't find any available server.", instanceId);
      217.                 });
      218.             }
      219.  
      220.             FAILED_COUNT = 0;
      221.         }
      222.         return null;
      223.     } else {
      224.         // 重置失败次数
      225.         FAILED_COUNT = 0;
      226.         log.debug("[PowerDiscovery] current server is {}.", result);
      227.         return result;
      228.     }
      229. }
      230.  
      231. /**
      232.  * 第192行代码处和第198行代码处
      233.  */
      234. @SuppressWarnings("rawtypes")
      235. private String acquire(String httpServerAddress) {
      236.     String result = null;
      237.     //构建url参数
      238.     String url = buildServerDiscoveryUrl(httpServerAddress);
      239.     try {
      240.         //请求服务端ServerController.acquireServer方法
      241.         result = CommonUtils.executeWithRetry0(() -> HttpUtils.get(url));
      242.     } catch (Exception ignore) {
      243.     }
      244.     if (!StringUtils.isEmpty(result)) {
      245.         try {
      246.             ResultDTO resultDTO = JsonUtils.parseObject(result, ResultDTO.class);
      247.             if (resultDTO.isSuccess()) {
      248.                 return resultDTO.getData().toString();
      249.             }
      250.         } catch (Exception ignore) {
      251.         }
      252.     }
      253.     return null;
      254. }
      255.  
      256. /**
      257.  * ServerController:
      258.  * 第241行代码处：
      259.  */
      260. @GetMapping("/acquire")
      261. public ResultDTO acquireServer(ServerDiscoveryRequest request) {
      262.     return ResultDTO.success(serverElectionService.elect(request));
      263. }
      264.  
      265. /**
      266.  * ServerElectionService:
      267.  */
      268. public String elect(ServerDiscoveryRequest request) {
      269.     if (!accurate()) {
      270.         final String currentServer = request.getCurrentServer();
      271.         // 如果是本机，就不需要查数据库那么复杂的操作了，直接返回成功
      272.         Optional localProtocolInfoOpt = Optional.ofNullable(transportService.allProtocols().get(request.getProtocol()));
      273.         //如果不是精确地请求，并且请求的参数中直接带有当前服务端的地址，则直接返回其地址即可
      274.         if (localProtocolInfoOpt.isPresent() && localProtocolInfoOpt.get().getAddress().equals(currentServer)) {
      275.             log.debug("[ServerElectionService] this server[{}] is worker's current server, skip check", currentServer);
      276.             return currentServer;
      277.         }
      278.     }
      279.     //上面的条件不满足，则走常规的选举流程
      280.     return getServer0(request);
      281. }
      282.  
      283. /**
      284.  * 第269行代码处：
      285.  */
      286. private boolean accurate() {
      287.     //accurateSelectServerPercentage默认为50，这里是在随机判断是否是精确的请求
      288.     return ThreadLocalRandom.current().nextInt(100) < accurateSelectServerPercentage;
      289. }
      290.  
      291. /**
      292.  * 第280行代码处：
      293.  */
      294. private String getServer0(ServerDiscoveryRequest discoveryRequest) {
      295.  
      296.     final Long appId = discoveryRequest.getAppId();
      297.     final String protocol = discoveryRequest.getProtocol();
      298.     Set downServerCache = Sets.newHashSet();
      299.  
      300.     for (int i = 0; i < RETRY_TIMES; i++) {
      301.  
      302.         // 无锁获取当前数据库中的Server
      303.         Optional appInfoOpt = appInfoRepository.findById(appId);
      304.         if (!appInfoOpt.isPresent()) {
      305.             throw new PowerJobException(appId + " is not registered!");
      306.         }
      307.         String appName = appInfoOpt.get().getAppName();
      308.         String originServer = appInfoOpt.get().getCurrentServer();
      309.         String activeAddress = activeAddress(originServer, downServerCache, protocol);
      310.         if (StringUtils.isNotEmpty(activeAddress)) {
      311.             return activeAddress;
      312.         }
      313.  
      314.         // 无可用Server，重新进行Server选举，需要加锁
      315.         String lockName = String.format(SERVER_ELECT_LOCK, appId);
      316.         //文件锁
      317.         boolean lockStatus = lockService.tryLock(lockName, 30000);
      318.         if (!lockStatus) {
      319.             try {
      320.                 Thread.sleep(500);
      321.             } catch (Exception ignore) {
      322.             }
      323.             continue;
      324.         }
      325.         try {
      326.  
      327.             // 可能上一台机器已经完成了Server选举，需要再次判断
      328.             //双重检查加锁
      329.             AppInfoDO appInfo = appInfoRepository.findById(appId).orElseThrow(() -> new RuntimeException("impossible, unless we just lost our database."));
      330.             String address = activeAddress(appInfo.getCurrentServer(), downServerCache, protocol);
      331.             if (StringUtils.isNotEmpty(address)) {
      332.                 return address;
      333.             }
      334.  
      335.             // 篡位，如果本机存在协议，则作为Server调度该 worker
      336.             //优先本机调度
      337.             final ProtocolInfo targetProtocolInfo = transportService.allProtocols().get(protocol);
      338.             if (targetProtocolInfo != null) {
      339.                 // 注意，写入 AppInfoDO#currentServer 的永远是 default 的地址，仅在返回的时候特殊处理为协议地址
      340.                 appInfo.setCurrentServer(transportService.defaultProtocol().getAddress());
      341.                 appInfo.setGmtModified(new Date());
      342.  
      343.                 appInfoRepository.saveAndFlush(appInfo);
      344.                 log.info("[ServerElection] this server({}) become the new server for app(appId={}).", appInfo.getCurrentServer(), appId);
      345.                 return targetProtocolInfo.getAddress();
      346.             }
      347.         } catch (Exception e) {
      348.             log.error("[ServerElection] write new server to db failed for app {}.", appName, e);
      349.         } finally {
      350.             lockService.unlock(lockName);
      351.         }
      352.     }
      353.     throw new PowerJobException("server elect failed for app " + appId);
      354. }
      355.  
      356. /**
      357.  * 第309行代码处：
      358.  * 判断指定server是否存活
      359.  *
      360.  * @param serverAddress   需要检测的server地址
      361.  * @param downServerCache 缓存，防止多次发送PING（这个QPS其实还蛮爆表的...）
      362.  * @param protocol        协议，用于返回指定的地址
      363.  * @return null or address
      364.  */
      365. private String activeAddress(String serverAddress, Set downServerCache, String protocol) {
      366.  
      367.     if (downServerCache.contains(serverAddress)) {
      368.         return null;
      369.     }
      370.     if (StringUtils.isEmpty(serverAddress)) {
      371.         return null;
      372.     }
      373.  
      374.     Ping ping = new Ping();
      375.     ping.setCurrentTime(System.currentTimeMillis());
      376.  
      377.     //构建服务端url参数
      378.     URL targetUrl = ServerURLFactory.ping2Friend(serverAddress);
      379.     try {
      380.         //这里会跳转到FriendActor.onReceivePing方法
      381.         AskResponse response = transportService.ask(Protocol.HTTP.name(), targetUrl, ping, AskResponse.class)
      382.                 .toCompletableFuture()
      383.                 .get(PING_TIMEOUT_MS, TimeUnit.MILLISECONDS);
      384.         if (response.isSuccess()) {
      385.             // 检测通过的是远程 server 的暴露地址，需要返回 worker 需要的协议地址
      386.             final JSONObject protocolInfo = JsonUtils.parseObject(response.getData(), JSONObject.class).getJSONObject(protocol);
      387.             if (protocolInfo != null) {
      388.                 downServerCache.remove(serverAddress);
      389.                 final String protocolAddress = protocolInfo.toJavaObject(ProtocolInfo.class).getAddress();
      390.                 log.info("[ServerElection] server[{}] is active, it will be the master, final protocol address={}", serverAddress, protocolAddress);
      391.                 return protocolAddress;
      392.             } else {
      393.                 log.warn("[ServerElection] server[{}] is active but don't have target protocol", serverAddress);
      394.             }
      395.         }
      396.     } catch (TimeoutException te) {
      397.         log.warn("[ServerElection] server[{}] was down due to ping timeout!", serverAddress);
      398.     } catch (Exception e) {
      399.         log.warn("[ServerElection] server[{}] was down with unknown case!", serverAddress, e);
      400.     }
      401.     downServerCache.add(serverAddress);
      402.     return null;
      403. }
      404.  
      405. /**
      406.  * FriendActor:
      407.  * 第381行代码处：
      408.  * 处理存活检测的请求
      409.  */
      410. @Handler(path = S4S_HANDLER_PING, processType = ProcessType.NO_BLOCKING)
      411. public AskResponse onReceivePing(Ping ping) {
      412.     return AskResponse.succeed(transportService.allProtocols());
      413. }
      414.  
      415. /**
      416.  * PowerTransportService：
      417.  */
      418. @Override
      419. public Map allProtocols() {
      420.     //这里就是从protocolName2Info缓存中取数据
      421.     return protocolName2Info;
      422. }
       

              由上可知，在客户端启动的时候，会通过选举的方式来选出一台服务器作为自己的server（也可能不选举，直接选择当前的服务端），并且会定时发送心跳数据来进行保活。        
       

              在上面第85行代码处和第86行代码处，客户端启动的时候也会有两个定时任务，查看其实现：
       

      3.1.1 WorkerHealthReporter
       

              WorkerHealthReporter是用来给客户端健康度定时上报用的：
       

      1. /**
      2.  * WorkerHealthReporter:
      3.  */
      4. @Override
      5. public void run() {
      6.  
      7.     // 没有可用Server，无法上报
      8.     String currentServer = workerRuntime.getServerDiscoveryService().getCurrentServerAddress();
      9.     if (StringUtils.isEmpty(currentServer)) {
      10.         log.warn("[WorkerHealthReporter] no available server,fail to report health info!");
      11.         return;
      12.     }
      13.  
      14.     SystemMetrics systemMetrics;
      15.  
      16.     if (workerRuntime.getWorkerConfig().getSystemMetricsCollector() == null) {
      17.         systemMetrics = SystemInfoUtils.getSystemMetrics();
      18.     } else {
      19.         systemMetrics = workerRuntime.getWorkerConfig().getSystemMetricsCollector().collect();
      20.     }
      21.  
      22.     WorkerHeartbeat heartbeat = new WorkerHeartbeat();
      23.  
      24.     heartbeat.setSystemMetrics(systemMetrics);
      25.     heartbeat.setWorkerAddress(workerRuntime.getWorkerAddress());
      26.     heartbeat.setAppName(workerRuntime.getWorkerConfig().getAppName());
      27.     heartbeat.setAppId(workerRuntime.getAppId());
      28.     heartbeat.setHeartbeatTime(System.currentTimeMillis());
      29.     heartbeat.setVersion(PowerJobWorkerVersion.getVersion());
      30.     heartbeat.setProtocol(workerRuntime.getWorkerConfig().getProtocol().name());
      31.     heartbeat.setClient("KingPenguin");
      32.     heartbeat.setTag(workerRuntime.getWorkerConfig().getTag());
      33.  
      34.     // 上报 Tracker 数量
      35.     heartbeat.setLightTaskTrackerNum(LightTaskTrackerManager.currentTaskTrackerSize());
      36.     heartbeat.setHeavyTaskTrackerNum(HeavyTaskTrackerManager.currentTaskTrackerSize());
      37.     // 是否超载
      38.     if (workerRuntime.getWorkerConfig().getMaxLightweightTaskNum() <= LightTaskTrackerManager.currentTaskTrackerSize() || workerRuntime.getWorkerConfig().getMaxHeavyweightTaskNum() <= HeavyTaskTrackerManager.currentTaskTrackerSize()) {
      39.         heartbeat.setOverload(true);
      40.     }
      41.     // 获取当前加载的容器列表
      42.     heartbeat.setContainerInfos(OmsContainerFactory.getDeployedContainerInfos());
      43.     // 发送请求
      44.     if (StringUtils.isEmpty(currentServer)) {
      45.         return;
      46.     }
      47.     // log
      48.     log.info("[WorkerHealthReporter] report health status,appId:{},appName:{},isOverload:{},maxLightweightTaskNum:{},currentLightweightTaskNum:{},maxHeavyweightTaskNum:{},currentHeavyweightTaskNum:{}",
      49.             heartbeat.getAppId(),
      50.             heartbeat.getAppName(),
      51.             heartbeat.isOverload(),
      52.             workerRuntime.getWorkerConfig().getMaxLightweightTaskNum(),
      53.             heartbeat.getLightTaskTrackerNum(),
      54.             workerRuntime.getWorkerConfig().getMaxHeavyweightTaskNum(),
      55.             heartbeat.getHeavyTaskTrackerNum()
      56.     );
      57.  
      58.     TransportUtils.reportWorkerHeartbeat(heartbeat, currentServer, workerRuntime.getTransporter());
      59. }
      60.  
      61. /**
      62.  * SystemInfoUtils:
      63.  * 第17行代码处：
      64.  */
      65. public static SystemMetrics getSystemMetrics() {
      66.  
      67.     SystemMetrics metrics = new SystemMetrics();
      68.  
      69.     //赋值cpu指标
      70.     fillCPUInfo(metrics);
      71.     //赋值内存指标
      72.     fillMemoryInfo(metrics);
      73.     //赋值磁盘指标
      74.     fillDiskInfo(metrics);
      75.  
      76.     // 在Worker完成分数计算，减小Server压力
      77.     metrics.calculateScore();
      78.     return metrics;
      79. }
      80.  
      81. /**
      82.  * TransportUtils:
      83.  * 第58行代码处：
      84.  */
      85. public static void reportWorkerHeartbeat(WorkerHeartbeat req, String address, Transporter transporter) {
      86.     //绑定url调用信息
      87.     final URL url = easyBuildUrl(ServerType.SERVER, S4W_PATH, S4W_HANDLER_WORKER_HEARTBEAT, address);
      88.     transporter.tell(url, req);
      89. }
       

              其中SystemInfoUtils.getSystemMetrics方法是在获取系统的一些指标数据（调用的都是Java的底层Api，这里就不再详细查看了），并且计算健康度的得分（之前在第2.3.1小节中已经查看过该方法的实现了）。基于此，我们就知道了控制台首页的worker指标数据是怎么来的了：
       

       

              在上面第88行代码处向服务端发送了心跳数据，接下来就来看下服务端是如何处理的：
       

      1. /**
      2.  * AbWorkerRequestHandler:
      3.  */
      4. @Override
      5. @Handler(path = S4W_HANDLER_WORKER_HEARTBEAT, processType = ProcessType.NO_BLOCKING)
      6. public void processWorkerHeartbeat(WorkerHeartbeat heartbeat) {
      7.     long startMs = System.currentTimeMillis();
      8.     WorkerHeartbeatEvent event = new WorkerHeartbeatEvent()
      9.             .setAppName(heartbeat.getAppName())
      10.             .setAppId(heartbeat.getAppId())
      11.             .setVersion(heartbeat.getVersion())
      12.             .setProtocol(heartbeat.getProtocol())
      13.             .setTag(heartbeat.getTag())
      14.             .setWorkerAddress(heartbeat.getWorkerAddress())
      15.             .setDelayMs(startMs - heartbeat.getHeartbeatTime())
      16.             .setScore(heartbeat.getSystemMetrics().getScore());
      17.     processWorkerHeartbeat0(heartbeat, event);
      18.     //默认实现是写入日志进行监控
      19.     monitorService.monitor(event);
      20. }
      21.  
      22. /**
      23.  * WorkerRequestHandlerImpl:
      24.  * 第17行代码处：
      25.  */
      26. @Override
      27. protected void processWorkerHeartbeat0(WorkerHeartbeat heartbeat, WorkerHeartbeatEvent event) {
      28.     WorkerClusterManagerService.updateStatus(heartbeat);
      29. }
      30.  
      31. /**
      32.  * WorkerClusterManagerService:
      33.  * 更新状态
      34.  *
      35.  * @param heartbeat Worker的心跳包
      36.  */
      37. public static void updateStatus(WorkerHeartbeat heartbeat) {
      38.     Long appId = heartbeat.getAppId();
      39.     String appName = heartbeat.getAppName();
      40.     ClusterStatusHolder clusterStatusHolder = APP_ID_2_CLUSTER_STATUS.computeIfAbsent(appId, ignore -> new ClusterStatusHolder(appName));
      41.     clusterStatusHolder.updateStatus(heartbeat);
      42. }
      43.  
      44. /**
      45.  * ClusterStatusHolder:
      46.  * 更新 worker 机器的状态
      47.  *
      48.  * @param heartbeat 心跳请求
      49.  */
      50. public void updateStatus(WorkerHeartbeat heartbeat) {
      51.  
      52.     String workerAddress = heartbeat.getWorkerAddress();
      53.     long heartbeatTime = heartbeat.getHeartbeatTime();
      54.  
      55.     WorkerInfo workerInfo = address2WorkerInfo.computeIfAbsent(workerAddress, ignore -> {
      56.         WorkerInfo wf = new WorkerInfo();
      57.         wf.refresh(heartbeat);
      58.         return wf;
      59.     });
      60.     long oldTime = workerInfo.getLastActiveTime();
      61.     //过期心跳数据，不处理
      62.     if (heartbeatTime < oldTime) {
      63.         log.warn("[ClusterStatusHolder-{}] receive the expired heartbeat from {}, serverTime: {}, heartTime: {}", appName, heartbeat.getWorkerAddress(), System.currentTimeMillis(), heartbeat.getHeartbeatTime());
      64.         return;
      65.     }
      66.  
      67.     workerInfo.refresh(heartbeat);
      68.  
      69.     List containerInfos = heartbeat.getContainerInfos();
      70.     if (!CollectionUtils.isEmpty(containerInfos)) {
      71.         containerInfos.forEach(containerInfo -> {
      72.             Map infos = containerId2Infos.computeIfAbsent(containerInfo.getContainerId(), ignore -> Maps.newConcurrentMap());
      73.             infos.put(workerAddress, containerInfo);
      74.         });
      75.     }
      76. }
      77.  
      78. /**
      79.  * WorkerInfo:
      80.  * 第57行代码处和第67行代码处：
      81.  * 刷新服务端记录的客户端的数据
      82.  */
      83. public void refresh(WorkerHeartbeat workerHeartbeat) {
      84.     address = workerHeartbeat.getWorkerAddress();
      85.     lastActiveTime = workerHeartbeat.getHeartbeatTime();
      86.     protocol = workerHeartbeat.getProtocol();
      87.     client = workerHeartbeat.getClient();
      88.     tag = workerHeartbeat.getTag();
      89.     systemMetrics = workerHeartbeat.getSystemMetrics();
      90.     containerInfos = workerHeartbeat.getContainerInfos();
      91.  
      92.     lightTaskTrackerNum = workerHeartbeat.getLightTaskTrackerNum();
      93.     heavyTaskTrackerNum = workerHeartbeat.getHeavyTaskTrackerNum();
      94.  
      95.     if (workerHeartbeat.isOverload()) {
      96.         overloading = true;
      97.         lastOverloadTime = workerHeartbeat.getHeartbeatTime();
      98.         log.warn("[WorkerInfo] worker {} is overload!", getAddress());
      99.     } else {
      100.         overloading = false;
      101.     }
      102. }
       

      3.1.2 LogSubmitter
       

      1. /**
      2.  * LogSubmitter:
      3.  */
      4. @Override
      5. public void run() {
      6.  
      7.     boolean lockResult = reportLock.tryLock();
      8.     if (!lockResult) {
      9.         return;
      10.     }
      11.  
      12.     try {
      13.  
      14.         final String currentServerAddress = serverDiscoveryService.getCurrentServerAddress();
      15.         // 当前无可用 Server
      16.         if (StringUtils.isEmpty(currentServerAddress)) {
      17.             if (!logQueue.isEmpty()) {
      18.                 logQueue.clear();
      19.                 log.warn("[OmsLogHandler] because there is no available server to report logs which leads to queue accumulation, oms discarded all logs.");
      20.             }
      21.             return;
      22.         }
      23.  
      24.         List logs = Lists.newLinkedList();
      25.  
      26.         while (!logQueue.isEmpty()) {
      27.             try {
      28.                 //从日志队列头部取数据
      29.                 InstanceLogContent logContent = logQueue.poll(100, TimeUnit.MILLISECONDS);
      30.                 logs.add(logContent);
      31.  
      32.                 //批处理
      33.                 if (logs.size() >= BATCH_SIZE) {
      34.                     WorkerLogReportReq req = new WorkerLogReportReq(workerAddress, Lists.newLinkedList(logs));
      35.                     // 不可靠请求，WEB日志不追求极致
      36.                     TransportUtils.reportLogs(req, currentServerAddress, transporter);
      37.                     logs.clear();
      38.                 }
      39.  
      40.             } catch (Exception ignore) {
      41.                 break;
      42.             }
      43.         }
      44.  
      45.         if (!logs.isEmpty()) {
      46.             WorkerLogReportReq req = new WorkerLogReportReq(workerAddress, logs);
      47.             TransportUtils.reportLogs(req, currentServerAddress, transporter);
      48.         }
      49.  
      50.     } finally {
      51.         reportLock.unlock();
      52.     }
      53. }
      54.  
      55. /**
      56.  * TransportUtils:
      57.  * 第36行代码处和第47行代码处：
      58.  */
      59. public static void reportLogs(WorkerLogReportReq req, String address, Transporter transporter) {
      60.     final URL url = easyBuildUrl(ServerType.SERVER, S4W_PATH, S4W_HANDLER_REPORT_LOG, address);
      61.     transporter.tell(url, req);
      62. }
       

              上面第61行代码处会将请求发送到服务端的AbWorkerRequestHandler.processWorkerLogReport方法中：
       

      1. /**
      2.  * AbWorkerRequestHandler:
      3.  */
      4. @Override
      5. @Handler(path = S4W_HANDLER_REPORT_LOG, processType = ProcessType.NO_BLOCKING)
      6. public void processWorkerLogReport(WorkerLogReportReq req) {
      7.  
      8.     WorkerLogReportEvent event = new WorkerLogReportEvent()
      9.             .setWorkerAddress(req.getWorkerAddress())
      10.             .setLogNum(req.getInstanceLogContents().size());
      11.     try {
      12.         processWorkerLogReport0(req, event);
      13.         event.setStatus(WorkerLogReportEvent.Status.SUCCESS);
      14.     } catch (RejectedExecutionException re) {
      15.         event.setStatus(WorkerLogReportEvent.Status.REJECTED);
      16.     } catch (Throwable t) {
      17.         event.setStatus(WorkerLogReportEvent.Status.EXCEPTION);
      18.         log.warn("[WorkerRequestHandler] process worker report failed!", t);
      19.     } finally {
      20.         //日志监控
      21.         monitorService.monitor(event);
      22.     }
      23. }
      24.  
      25. /**
      26.  * WorkerRequestHandlerImpl:
      27.  * 第12行代码处：
      28.  */
      29. @Override
      30. protected void processWorkerLogReport0(WorkerLogReportReq req, WorkerLogReportEvent event) {
      31.     // 这个效率应该不会拉垮吧...也就是一些判断 + Map#get 吧...
      32.     instanceLogService.submitLogs(req.getWorkerAddress(), req.getInstanceLogContents());
      33. }
      34.  
      35. /**
      36.  * InstanceLogService:
      37.  * 提交日志记录，持久化到本地数据库中
      38.  *
      39.  * @param workerAddress 上报机器地址
      40.  * @param logs          任务实例运行时日志
      41.  */
      42. @Async(value = PJThreadPool.LOCAL_DB_POOL)
      43. public void submitLogs(String workerAddress, List logs) {
      44.  
      45.     List logList = logs.stream().map(x -> {
      46.         instanceId2LastReportTime.put(x.getInstanceId(), System.currentTimeMillis());
      47.  
      48.         LocalInstanceLogDO y = new LocalInstanceLogDO();
      49.         BeanUtils.copyProperties(x, y);
      50.         y.setWorkerAddress(workerAddress);
      51.         return y;
      52.     }).collect(Collectors.toList());
      53.  
      54.     try {
      55.         //插入到local_instance_log表中
      56.         CommonUtils.executeWithRetry0(() -> localInstanceLogRepository.saveAll(logList));
      57.     } catch (Exception e) {
      58.         log.warn("[InstanceLogService] persistent instance logs failed, these logs will be dropped: {}.", logs, e);
      59.     }
      60. }
       

       

       
 
      原创不易，未得准许，请勿转载，翻版必究
       

      
                
                    
                

                
    190. 

                    相关阅读:
      
                    
Java版工程行业管理系统源码-专业的工程管理软件- 工程项目各模块及其功能点清单                            
      
javaee SpringMVC 乱码问题解决                            
      
docker如何查看对外暴露接口                            
      
人工智能：人脸识别技术应用场景介绍                            
      
20.1CubeMx配置FMC控制SDRAM【W9825G6KH-6】                            
      
day-45 代码随想录算法训练营（19）动态规划 part 07                            
      
SpringBoot自动配置原理解析 | 京东物流技术团队                            
      
static关键字修饰成员变量与成员函数                            
      
CI/CD持续集成/持续部署                            
      
超详细Redis使用手册                            
      
                          
                
      191. 
                
    191. 
                    原文地址：https://blog.csdn.net/weixin_30342639/article/details/132662425
                
      192.