devbox
从前慢现在也慢
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

LTS原理--TaskTracker任务处理（四）

作者：从前慢现在也慢 | 2024-05-06 15:20:55

赞

踩

tasktracker

TaskTracker主要的功能是执行任务，其有两点操作：

（1）当TaskTracker启动时会根据当前TaskTracker资源定时向JobTracker

（2）TaskTracker接收到JobTracker推送的任务执行任务。

参考LTS原理--JobTracker任务接收与分配（三）

一、示例：

地址：https://github.com/ltsopensource/lts-examples/tree/master/lts-example-tasktracker

1、配置

  1.   <context:component-scan base-package="com.github.ltsopensource.example"/>
  2.  
  3.     <bean id="taskTracker" class="com.github.ltsopensource.spring.TaskTrackerAnnotationFactoryBean" init-method="start">
  4.         //任务执行类
  5.         <property name="jobRunnerClass" value="com.github.ltsopensource.example.spring.SpringAnnotationJobRunner"/>
  6.         <property name="clusterName" value="test_cluster"/>
  7.         <property name="registryAddress" value="zookeeper://127.0.0.1:2181"/>
  8.         <property name="nodeGroup" value="test_trade_TaskTracker"/>
  9.         <property name="workThreads" value="64"/>
  10.         <property name="configs">
  11.             <props>
  12.                 <prop key="job.fail.store">mapdb</prop>
  13.             </props>
  14.         </property>
  15.     </bean>

2、任务执行类实现JobRunner接口

  1. public class SpringAnnotationJobRunner implements JobRunner {
  2.  
  3.     private static final Logger LOGGER = LoggerFactory.getLogger(SpringAnnotationJobRunner.class);
  4.  
  5.     @Autowired
  6.     SpringBean springBean;
  7.  
  8.     @Override
  9.     public Result run(JobContext jobContext) throws Throwable {
  10.         try {
  11.             Thread.sleep(1000L);
  12.  
  13.             springBean.hello();
  14.  
  15.             // TODO 业务逻辑
  16.             LOGGER.info("我要执行：" + jobContext.getJob());
  17.             BizLogger bizLogger = jobContext.getBizLogger();
  18.             // 会发送到 LTS (JobTracker上)
  19.             bizLogger.info("测试，业务日志啊啊啊啊啊");
  20.  
  21.         } catch (Exception e) {
  22.             LOGGER.info("Run job failed!", e);
  23.             return new Result(Action.EXECUTE_LATER, e.getMessage());
  24.         }
  25.         return new Result(Action.EXECUTE_SUCCESS, "执行成功了，哈哈");
  26.     }
  27.  
  28. }

二、请求任务

在TaskTracker初始化时会启用定时任务，根据服务器线程资源请求向JobTracker发送获取任务的请求。

在TaskTracker的beforeStart中初始化时会初始化JobPullMachine会建立定时任务向JobTracker发送任务请求。

  1.  @Override
  2.     protected void beforeStart() {
  3.         appContext.setMStatReporter(new TaskTrackerMStatReporter(appContext));
  4.  
  5.         appContext.setRemotingClient(remotingClient);
  6.         // 设置 线程池
  7.         appContext.setRunnerPool(new RunnerPool(appContext));
  8.         appContext.getMStatReporter().start();
  9.         //拉取任务
  10.         appContext.setJobPullMachine(new JobPullMachine(appContext));
  11.         appContext.setStopWorkingMonitor(new StopWorkingMonitor(appContext));
  12.  
  13.         appContext.getHttpCmdServer().registerCommands(
  14.                 new JobTerminateCmd(appContext));     // 终止某个正在执行的任务
  15.     }

在JobPullMachine初始化时会监听注册中心中JobTracker节点的信息，如果存在这个节点，则会调用start方法，start方法中建立定时任务向JobTracker发送请求。

  1.  public JobPullMachine(final TaskTrackerAppContext appContext) {
  2.         this.appContext = appContext;
  3.         this.jobPullFrequency = appContext.getConfig().getParameter(ExtConfig.JOB_PULL_FREQUENCY, Constants.DEFAULT_JOB_PULL_FREQUENCY);
  4.  
  5.         this.machineResCheckEnable = appContext.getConfig().getParameter(ExtConfig.LB_MACHINE_RES_CHECK_ENABLE, false);
  6. 		//从zk中监听JobTracker节点，如果存在节点则调用start方法，start方法中会建立定时任务执行worker线程
  7.         appContext.getEventCenter().subscribe(
  8.                 new EventSubscriber(JobPullMachine.class.getSimpleName().concat(appContext.getConfig().getIdentity()),
  9.                         new Observer() {
  10.                             @Override
  11.                             public void onObserved(EventInfo eventInfo) {
  12.                                 if (EcTopic.JOB_TRACKER_AVAILABLE.equals(eventInfo.getTopic())) {
  13.                                     // JobTracker 可用了
  14.                                     start();
  15.                                 } else if (EcTopic.NO_JOB_TRACKER_AVAILABLE.equals(eventInfo.getTopic())) {
  16.                                     stop();
  17.                                 }
  18.                             }
  19.                         }), EcTopic.JOB_TRACKER_AVAILABLE, EcTopic.NO_JOB_TRACKER_AVAILABLE);
  20.         this.worker = new Runnable() {
  21.             @Override
  22.             public void run() {
  23.                 try {
  24.                     if (!start.get()) {
  25.                         return;
  26.                     }
  27.                     if (!isMachineResEnough()) {
  28.                         // 如果机器资源不足,那么不去取任务
  29.                         return;
  30.                     }
  31.                     //向JobTracker服务器发送任务请求
  32.                     sendRequest();
  33.                 } catch (Exception e) {
  34.                     LOGGER.error("Job pull machine run error!", e);
  35.                 }
  36.             }
  37.         };
  38.     }

在start方法中建立定时任务，定时任务中会调用sendRequest方法，定时向JobTracker服务发送任务请求。

  1. private void start() {
  2.         try {
  3.             if (start.compareAndSet(false, true)) {
  4.                 if (scheduledFuture == null) {
  5.                     scheduledFuture = executorService.scheduleWithFixedDelay(worker, jobPullFrequency * 1000, jobPullFrequency * 1000, TimeUnit.MILLISECONDS);
  6.                 }
  7.                 LOGGER.info("Start Job pull machine success!");
  8.             }
  9.         } catch (Throwable t) {
  10.             LOGGER.error("Start Job pull machine failed!", t);
  11.         }
  12.     }

在sendRequest中会判断当前TaskTracker可执行线程数，然后向JobTracker发送请求

  1. private void sendRequest() throws RemotingCommandFieldCheckException {
  2. 		//判断当前节点还有可执行线程数
  3.         int availableThreads = appContext.getRunnerPool().getAvailablePoolSize();
  4.         if (LOGGER.isDebugEnabled()) {
  5.             LOGGER.debug("current availableThreads:{}", availableThreads);
  6.         }
  7.         if (availableThreads == 0) {
  8.             return;
  9.         }
  10. 		//创建请求体
  11.         JobPullRequest requestBody = appContext.getCommandBodyWrapper().wrapper(new JobPullRequest());
  12.         requestBody.setAvailableThreads(availableThreads);
  13.         RemotingCommand request = RemotingCommand.createRequestCommand(JobProtos.RequestCode.JOB_PULL.code(), requestBody);
  14.  
  15.         try {
  16. 			//向JobTracker发送任务请求
  17.             RemotingCommand responseCommand = appContext.getRemotingClient().invokeSync(request);
  18.             if (responseCommand == null) {
  19.                 LOGGER.warn("Job pull request failed! response command is null!");
  20.                 return;
  21.             }
  22.             if (JobProtos.ResponseCode.JOB_PULL_SUCCESS.code() == responseCommand.getCode()) {
  23.                 if (LOGGER.isDebugEnabled()) {
  24.                     LOGGER.debug("Job pull request success!");
  25.                 }
  26.                 return;
  27.             }
  28.             LOGGER.warn("Job pull request failed! response command is null!");
  29.         } catch (JobTrackerNotFoundException e) {
  30.             LOGGER.warn("no job tracker available!");
  31.         }
  32.     }

在RemotingClientDelegate的invokeSync方法中会从注册中心获取JobTracker节点信息，然后建立远程连接发送请求。

  1. /**
  2.      * 同步调用
  3.      */
  4.     public RemotingCommand invokeSync(RemotingCommand request)
  5.             throws JobTrackerNotFoundException {
  6.  
  7.         Node jobTracker = getJobTrackerNode();
  8.  
  9.         try {
  10.             RemotingCommand response = remotingClient.invokeSync(jobTracker.getAddress(),
  11.                     request, appContext.getConfig().getInvokeTimeoutMillis());
  12.             this.serverEnable = true;
  13.             return response;
  14.         } catch (Exception e) {
  15.             // 将这个JobTracker移除
  16.             jobTrackers.remove(jobTracker);
  17.             try {
  18.                 Thread.sleep(100L);
  19.             } catch (InterruptedException e1) {
  20.                 LOGGER.error(e1.getMessage(), e1);
  21.             }
  22.             // 只要不是节点 不可用, 轮询所有节点请求
  23.             return invokeSync(request);
  24.         }
  25.     }

三、接收任务执行

TaskTracker也是提供了RemotingDispatcher类，用来接收JobTracker发送过来的任务

  1. public class RemotingDispatcher extends AbstractProcessor {
  2.  
  3.     private final Map<JobProtos.RequestCode, RemotingProcessor> processors = new HashMap<JobProtos.RequestCode, RemotingProcessor>();
  4.  
  5.     public RemotingDispatcher(TaskTrackerAppContext appContext) {
  6.         super(appContext);
  7.         processors.put(JobProtos.RequestCode.PUSH_JOB, new JobPushProcessor(appContext));
  8.         processors.put(JobProtos.RequestCode.JOB_ASK, new JobAskProcessor(appContext));
  9.     }
  10.     //处理JobTracker发送过来的任务
  11.     @Override
  12.     public RemotingCommand processRequest(Channel channel, RemotingCommand request) throws RemotingCommandException {
  13.  
  14.         JobProtos.RequestCode code = JobProtos.RequestCode.valueOf(request.getCode());
  15.         RemotingProcessor processor = processors.get(code);
  16.         if (processor == null) {
  17.             return RemotingCommand.createResponseCommand(RemotingProtos.ResponseCode.REQUEST_CODE_NOT_SUPPORTED.code(),
  18.                     "request code not supported!");
  19.         }
  20.         return processor.processRequest(channel, request);
  21.     }
  22.  
  23. }

在JobPushProcessor的processRequest中会根据JobTracker发送过来的数据，线程池中执行相关任务。

  1. @Override
  2.     public RemotingCommand processRequest(Channel channel,
  3.                                           final RemotingCommand request) throws RemotingCommandException {
  4.  
  5. 		//获取请求数据
  6.         JobPushRequest requestBody = request.getBody();
  7.  
  8.         // JobTracker 分发来的 job
  9.         final List<JobMeta> jobMetaList = requestBody.getJobMetaList();
  10.         List<String> failedJobIds = null;
  11.  
  12.         for (JobMeta jobMeta : jobMetaList) {
  13.             try {
  14. 				//执行任务
  15.                 appContext.getRunnerPool().execute(jobMeta, jobRunnerCallback);
  16.             } catch (NoAvailableJobRunnerException e) {
  17.                 if (failedJobIds == null) {
  18.                     failedJobIds = new ArrayList<String>();
  19.                 }
  20.                 failedJobIds.add(jobMeta.getJobId());
  21.             }
  22.         }
  23.         if (CollectionUtils.isNotEmpty(failedJobIds)) {
  24.             // 任务推送失败
  25.             JobPushResponse jobPushResponse = new JobPushResponse();
  26.             jobPushResponse.setFailedJobIds(failedJobIds);
  27.             return RemotingCommand.createResponseCommand(JobProtos.ResponseCode.NO_AVAILABLE_JOB_RUNNER.code(), jobPushResponse);
  28.         }
  29.  
  30.         // 任务推送成功
  31.         return RemotingCommand.createResponseCommand(JobProtos
  32.                 .ResponseCode.JOB_PUSH_SUCCESS.code(), "job push success!");
  33.     }

在RunnerPool中调用execute执行任务，会创建线程在线程池中进行执行

  1. public void execute(JobMeta jobMeta, RunnerCallback callback) throws NoAvailableJobRunnerException {
  2.         try {
  3.             threadPoolExecutor.execute(
  4.                     new JobRunnerDelegate(appContext, jobMeta, callback));
  5.             if (LOGGER.isDebugEnabled()) {
  6.                 LOGGER.debug("Receive job success ! " + jobMeta);
  7.             }
  8.         } catch (RejectedExecutionException e) {
  9.             LOGGER.warn("No more thread to run job .");
  10.             throw new NoAvailableJobRunnerException(e);
  11.         }
  12.     }

在线程JobRunnerDelegate中调用run方法执行，通过反射调用jobRunnerClass的实现类，完成run执行

  1. @Override
  2.     public void run() {
  3.  
  4.         thread = Thread.currentThread();
  5.  
  6.         try {
  7.             blockedOn(interruptor);
  8.             if (Thread.currentThread().isInterrupted()) {
  9.                 ((InterruptibleAdapter) interruptor).interrupt();
  10.             }
  11.  
  12.             while (jobMeta != null) {
  13.                 long startTime = SystemClock.now();
  14.                 // 设置当前context中的jobId
  15.                 Response response = new Response();
  16.                 response.setJobMeta(jobMeta);
  17.  
  18.                 BizLoggerAdapter logger = (BizLoggerAdapter) BizLoggerFactory.getLogger(
  19.                         appContext.getBizLogLevel(),
  20.                         appContext.getRemotingClient(), appContext);
  21.  
  22.                 try {
  23.                     appContext.getRunnerPool().getRunningJobManager()
  24.                             .in(jobMeta.getJobId(), this);
  25. 					//获取JobRunner接口的实现类
  26.                     this.curJobRunner = appContext.getRunnerPool().getRunnerFactory().newRunner();
  27. 					//调用run方法，完成任务调度
  28.                     Result result = this.curJobRunner.run(buildJobContext(logger, jobMeta));
  29.  
  30.                     if (result == null) {
  31.                         response.setAction(Action.EXECUTE_SUCCESS);
  32.                     } else {
  33.                         if (result.getAction() == null) {
  34.                             response.setAction(Action.EXECUTE_SUCCESS);
  35.                         } else {
  36.                             response.setAction(result.getAction());
  37.                         }
  38.                         response.setMsg(result.getMsg());
  39.                     }
  40.  
  41.                     long time = SystemClock.now() - startTime;
  42.                     stat.addRunningTime(time);
  43.                     if (LOGGER.isDebugEnabled()) {
  44.                         LOGGER.debug("Job execute completed : {}, time:{} ms.", jobMeta.getJob(), time);
  45.                     }
  46.                 } catch (Throwable t) {
  47.                     StringWriter sw = new StringWriter();
  48.                     t.printStackTrace(new PrintWriter(sw));
  49.                     response.setAction(Action.EXECUTE_EXCEPTION);
  50.                     response.setMsg(sw.toString());
  51.                     long time = SystemClock.now() - startTime;
  52.                     stat.addRunningTime(time);
  53.                     LOGGER.error("Job execute error : {}, time: {}, {}", jobMeta.getJob(), time, t.getMessage(), t);
  54.                 } finally {
  55.                     checkInterrupted(logger);
  56.                     appContext.getRunnerPool().getRunningJobManager()
  57.                             .out(jobMeta.getJobId());
  58.                 }
  59.                 // 统计数据
  60.                 stat(response.getAction());
  61.  
  62.                 if (isStopToGetNewJob()) {
  63.                     response.setReceiveNewJob(false);
  64.                 }
  65.                 this.jobMeta = callback.runComplete(response);
  66.                 DotLogUtils.dot("JobRunnerDelegate.run get job " + (this.jobMeta == null ? "NULL" : "NOT_NULL"));
  67.             }
  68.         } finally {
  69.             blockedOn(null);
  70.         }
  71.     }

总结：

TaskTracker主要有两方面：

（1）主动向JobTracker发送请求获取任务

（2）监听JobTracker发送过来的任务进行任务执行

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/从前慢现在也慢/article/detail/544803
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号