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

古城微笑少年丶 2022-05-11 10:33 367阅读 0赞

JobTracker在这做的是类似路由的作用，JobClient将任务信息提交到JobTracker，JobTracker将消息信息路由分发给TaskTracker进行消息处理。

在上一篇博客[LTS原理--JobClient提交任务过程（二）][LTS_--JobClient]中我们已经介绍了JobClient的任务提交，接下来我们看看在JobTracker中对接收到的任务信息做了什么处理操作。

### 一、示例 ###

1、JobTracker服务注册

（1）zk服务注册地址

（2）对外服务端口30005

（3）数据相关配置

<bean id="jobTracker" class="com.github.ltsopensource.spring.JobTrackerFactoryBean" init-method="start">
            <property name="clusterName" value="test_cluster"/>
            <property name="registryAddress" value="zookeeper://127.0.0.1:2181"/>
            <property name="listenPort" value="30005"/>
            <property name="configs">
                <props>
                    <prop key="job.logger">mysql</prop>
                    <prop key="job.queue">mysql</prop>
                    <prop key="jdbc.url">jdbc:mysql://127.0.0.1:3306/lts</prop>
                    <prop key="jdbc.username">root</prop>
                    <prop key="jdbc.password">root</prop>
                </props>
            </property>
        </bean>

（2）Main函数

public class Main {
    
        public static void main(String[] args) {
            ApplicationContext context = new ClassPathXmlApplicationContext("/lts-jobtracker.xml");
        }
    
    }

在JobTrackerFactoryBean进行初始化时会将服务注册到zk，对外提供30005服务端口及连接数据库操作。

### 二、任务接收 ###

JobTracker提供了一个RemotingDispatcher请求接收分发器，当JobClient提交任务时，会由JobSubmitProcessor进行处理，将接收到的消息添加到一个消息队列中。

由于LTS封装了Netty、Mina和LTS通信框架，最最终消息处理还是调用RemotingDispatcher的processRequest方法，接下来我们看看做了什么处理。

在doBiz中进行最终业务进行处理。

@Override
        public RemotingCommand processRequest(Channel channel, RemotingCommand request) throws RemotingCommandException {
            //真的不同的消息进行不同处理
            // 心跳
            if (request.getCode() == JobProtos.RequestCode.HEART_BEAT.code()) {
                offerHandler(channel, request);
                return RemotingCommand.createResponseCommand(JobProtos.ResponseCode.HEART_BEAT_SUCCESS.code(), "");
            }
            //限流请求
            if (reqLimitEnable) {
                return doBizWithReqLimit(channel, request);
            } else {
            //处理具体的业务
                return doBiz(channel, request);
            }
        }

doBiz中会根据不同的RequestCode选择不同的处理器RemotingProcessor进行处理，当JobClient提交任务时，此时RemotingProcessor的实现类JobSubmitProcessor对请求进行处理。

private RemotingCommand doBiz(Channel channel, RemotingCommand request) throws RemotingCommandException {
            // 其他的请求code
            RequestCode code = RequestCode.valueOf(request.getCode());
            //根据不同的请求选择不同的请求处理器
            RemotingProcessor processor = processors.get(code);
            if (processor == null) {
                return RemotingCommand.createResponseCommand(RemotingProtos.ResponseCode.REQUEST_CODE_NOT_SUPPORTED.code(), "request code not supported!");
            }
            offerHandler(channel, request);
            return processor.processRequest(channel, request);
        }

在JobSubmitProcessor的processRequest中就是将任务信息添加到消息队列中，并且最终消息会进行入库处理，然后创建返回Response。

@Override
        public RemotingCommand processRequest(Channel channel, RemotingCommand request) throws RemotingCommandException {
    
            JobSubmitRequest jobSubmitRequest = request.getBody();
    
            JobSubmitResponse jobSubmitResponse = appContext.getCommandBodyWrapper().wrapper(new JobSubmitResponse());
            RemotingCommand response;
            try {
                //将消息添加到消息队列中
                appContext.getJobReceiver().receive(jobSubmitRequest);
                //返回任务提交成功
                response = RemotingCommand.createResponseCommand(
                        JobProtos.ResponseCode.JOB_RECEIVE_SUCCESS.code(), "job submit success!", jobSubmitResponse);
    
            } catch (JobReceiveException e) {
                LOGGER.error("Receive job failed , jobs = " + jobSubmitRequest.getJobs(), e);
                jobSubmitResponse.setSuccess(false);
                jobSubmitResponse.setMsg(e.getMessage());
                jobSubmitResponse.setFailedJobs(e.getJobs());
                response = RemotingCommand.createResponseCommand(
                        JobProtos.ResponseCode.JOB_RECEIVE_FAILED.code(), e.getMessage(), jobSubmitResponse);
            }
    
            return response;
        }

在JobReceive的receive方法中会进行消息接收操作，最终将消息入库

public void receive(JobSubmitRequest request) throws JobReceiveException {
    
            List<Job> jobs = request.getJobs();
            if (CollectionUtils.isEmpty(jobs)) {
                return;
            }
            JobReceiveException exception = null;
            for (Job job : jobs) {
                try {
                    //消息添加到队列
                    addToQueue(job, request);
                } catch (Exception e) {
                    if (exception == null) {
                        exception = new JobReceiveException(e);
                    }
                    exception.addJob(job);
                }
            }
    
            if (exception != null) {
                throw exception;
            }
        }

在addToQueue中将消息添加并记录日志

private JobPo addToQueue(Job job, JobSubmitRequest request) {
    
            JobPo jobPo = null;
            boolean success = false;
            BizLogCode code = null;
            try {
                jobPo = JobDomainConverter.convert(job);
                if (jobPo == null) {
                    LOGGER.warn("Job can not be null。{}", job);
                    return null;
                }
                if (StringUtils.isEmpty(jobPo.getSubmitNodeGroup())) {
                    jobPo.setSubmitNodeGroup(request.getNodeGroup());
                }
                // 设置 jobId
                jobPo.setJobId(JobUtils.generateJobId());
    
                // 添加任务
                addJob(job, jobPo);
    
                success = true;
                code = BizLogCode.SUCCESS;
    
            } catch (DupEntryException e) {
                // 已经存在
                if (job.isReplaceOnExist()) {
                    Assert.notNull(jobPo);
                    success = replaceOnExist(job, jobPo);
                    code = success ? BizLogCode.DUP_REPLACE : BizLogCode.DUP_FAILED;
                } else {
                    code = BizLogCode.DUP_IGNORE;
                    LOGGER.info("Job already exist And ignore. nodeGroup={}, {}", request.getNodeGroup(), job);
                }
            } finally {
                if (success) {
                    stat.incReceiveJobNum();
                    if (LOGGER.isDebugEnabled()) {
                        LOGGER.debug("Receive Job success. {}", job);
                    }
                }
            }
    
            // 记录日志
            jobBizLog(jobPo, code);
    
            return jobPo;
        }

根据不同的消息类型添加消息到数据库中

/**
         * 添加任务
         */
        private void addJob(Job job, JobPo jobPo) throws DupEntryException {
            if (job.isCron()) {
                addCronJob(jobPo);
            } else if (job.isRepeatable()) {
                addRepeatJob(jobPo);
            } else {
                addTriggerTimeJob(jobPo);
            }
        }

### 三、任务发送 ###

TaskTracker通过拉去的方式从JobTracker获取任务，接下来我们看看JobTracker发送任务是做了什么操作。

在RemotingDispatcher中doBiz根据code类型来选择处理器RemotingProcessor

private RemotingCommand doBiz(Channel channel, RemotingCommand request) throws RemotingCommandException {
            // 其他的请求code
            RequestCode code = RequestCode.valueOf(request.getCode());
            //根据code选择处理器
            RemotingProcessor processor = processors.get(code);
            if (processor == null) {
                return RemotingCommand.createResponseCommand(RemotingProtos.ResponseCode.REQUEST_CODE_NOT_SUPPORTED.code(), "request code not supported!");
            }
            offerHandler(channel, request);
            return processor.processRequest(channel, request);
        }

在JobPullProcessor中processRequest会根据请求内容，选择将任务分发到对应到TaskTracker中。

@Override
        public RemotingCommand processRequest(final Channel ctx, final RemotingCommand request) throws RemotingCommandException {
            //获取请求体
            JobPullRequest requestBody = request.getBody();
    
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("taskTrackerNodeGroup:{}, taskTrackerIdentity:{} , availableThreads:{}", requestBody.getNodeGroup(), requestBody.getIdentity(), requestBody.getAvailableThreads());
            }
            //添加到处理job中
            jobPusher.push(requestBody);
    
            return RemotingCommand.createResponseCommand(JobProtos.ResponseCode.JOB_PULL_SUCCESS.code(), "");
        }

在JobPusher的push方法中会新起线程处理TaskTracker发送过来的任务请求。

public void push(final JobPullRequest request) {
    
            this.executorService.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        //新起线程处理请求
                        push0(request);
                    } catch (Exception e) {
                        LOGGER.error("Job push failed!", e);
                    }
                }
            });
        }

在push0中就是根据TaskTracker的可用工作线程数，来推送对应的数量的任务

private void push0(final JobPullRequest request) {
    
    		//获取分组信息
            String nodeGroup = request.getNodeGroup();
            String identity = request.getIdentity();
            // 更新TaskTracker的可用线程数
            appContext.getTaskTrackerManager().updateTaskTrackerAvailableThreads(nodeGroup,
                    identity, request.getAvailableThreads(), request.getTimestamp());
    
    		//获取TaskTracker节点信息
            final TaskTrackerNode taskTrackerNode = appContext.getTaskTrackerManager().
                    getTaskTrackerNode(nodeGroup, identity);
    
            if (taskTrackerNode == null) {
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("taskTrackerNodeGroup:{}, taskTrackerIdentity:{} , didn't have node.", nodeGroup, identity);
                }
                return;
            }
    
            int availableThread = taskTrackerNode.getAvailableThread().get();
            if (availableThread <= 0) {
                return;
            }
    
            AtomicBoolean pushingFlag = getPushingFlag(taskTrackerNode);
            if (pushingFlag.compareAndSet(false, true)) {
                try {
                    final int batchSize = jobPushBatchSize;
    
                    int it = availableThread % batchSize == 0 ? availableThread / batchSize : availableThread / batchSize + 1;
    
                    final CountDownLatch latch = new CountDownLatch(it);
    
                    for (int i = 1; i <= it; i++) {
                        int size = batchSize;
                        if (i == it) {
                            size = availableThread - batchSize * (it - 1);
                        }
                        final int finalSize = size;
    					//创建多个线程去推送任务
                        pushExecutorService.execute(new Runnable() {
                            @Override
                            public void run() {
                                try {
                                    // 推送任务
                                    send(remotingServer, finalSize, taskTrackerNode);
                                } catch (Throwable t) {
                                    LOGGER.error("Error on Push Job to {}", taskTrackerNode, t);
                                } finally {
                                    latch.countDown();
                                }
                            }
                        });
                    }
    
                    try {
                        latch.await();
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    DotLogUtils.dot("taskTrackerNodeGroup:{}, taskTrackerIdentity:{} , pushing finished. batchTimes:{}, size:{}", nodeGroup, identity, it, availableThread);
                } finally {
                    pushingFlag.compareAndSet(true, false);
                }
            }
        }

在send方法中，会查询未处理的任务，将任务信息推送到TaskTracker中

private JobPushResult send(final RemotingServerDelegate remotingServer, int size, final TaskTrackerNode taskTrackerNode) {
    
    		//任务分组
            final String nodeGroup = taskTrackerNode.getNodeGroup();
            final String identity = taskTrackerNode.getIdentity();
    		//推送任务
            JobSender.SendResult sendResult = appContext.getJobSender().send(nodeGroup, identity, size, new JobSender.SendInvoker() {
                @Override
                public JobSender.SendResult invoke(final List<JobPo> jobPos) {
    
                    // 发送给TaskTracker执行
                    JobPushRequest body = appContext.getCommandBodyWrapper().wrapper(new JobPushRequest());
                    body.setJobMetaList(JobDomainConverter.convert(jobPos));
                    RemotingCommand commandRequest = RemotingCommand.createRequestCommand(JobProtos.RequestCode.PUSH_JOB.code(), body);
    
                    // 是否分发推送任务成功
                    final Holder<Boolean> pushSuccess = new Holder<Boolean>(false);
    
                    final CountDownLatch latch = new CountDownLatch(1);
                    try {
    					//建立连接发送消息
                        remotingServer.invokeAsync(taskTrackerNode.getChannel().getChannel(), commandRequest, new AsyncCallback() {
                            @Override
                            public void operationComplete(ResponseFuture responseFuture) {
                                try {
                                    RemotingCommand responseCommand = responseFuture.getResponseCommand();
                                    if (responseCommand == null) {
                                        LOGGER.warn("Job push failed! response command is null!");
                                        return;
                                    }
                                    if (responseCommand.getCode() == JobProtos.ResponseCode.JOB_PUSH_SUCCESS.code()) {
                                        if (LOGGER.isDebugEnabled()) {
                                            LOGGER.debug("Job push success! nodeGroup=" + nodeGroup + ", identity=" + identity + ", jobList=" + JSON.toJSONString(jobPos));
                                        }
                                        pushSuccess.set(true);
                                        stat.incPushJobNum(jobPos.size());
                                    } else if (responseCommand.getCode() == JobProtos.ResponseCode.NO_AVAILABLE_JOB_RUNNER.code()) {
                                        JobPushResponse jobPushResponse = responseCommand.getBody();
                                        if (jobPushResponse != null && CollectionUtils.isNotEmpty(jobPushResponse.getFailedJobIds())) {
                                            // 修复任务
                                            for (String jobId : jobPushResponse.getFailedJobIds()) {
                                                for (JobPo jobPo : jobPos) {
                                                    if (jobId.equals(jobPo.getJobId())) {
                                                        resumeJob(jobPo);
                                                        break;
                                                    }
                                                }
                                            }
                                            stat.incPushJobNum(jobPos.size() - jobPushResponse.getFailedJobIds().size());
                                        } else {
                                            stat.incPushJobNum(jobPos.size());
                                        }
                                        pushSuccess.set(true);
                                    }
    
                                } finally {
                                    latch.countDown();
                                }
                            }
                        });
    
                    } catch (RemotingSendException e) {
                        LOGGER.error("Remoting send error, jobPos={}", JSON.toJSONObject(jobPos), e);
                        return new JobSender.SendResult(false, JobPushResult.SENT_ERROR);
                    }
    
                    try {
                        latch.await(Constants.LATCH_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                        throw new RequestTimeoutException(e);
                    }
    
                    if (!pushSuccess.get()) {
                        if (LOGGER.isDebugEnabled()) {
                            LOGGER.debug("Job push failed! nodeGroup=" + nodeGroup + ", identity=" + identity + ", jobs=" + JSON.toJSONObject(jobPos));
                        }
                        for (JobPo jobPo : jobPos) {
                            resumeJob(jobPo);
                        }
                        return new JobSender.SendResult(false, JobPushResult.SENT_ERROR);
                    }
    
                    return new JobSender.SendResult(true, JobPushResult.SUCCESS);
                }
            });
    
            return (JobPushResult) sendResult.getReturnValue();
        }

在JobSender中调用send方法，完成任务推送处理。

public SendResult send(String taskTrackerNodeGroup, String taskTrackerIdentity, int size, SendInvoker invoker) {
    
    		//根据任务分钟获取任务
            List<JobPo> jobPos = fetchJob(taskTrackerNodeGroup, taskTrackerIdentity, size);
            if (jobPos.size() == 0) {
                return new SendResult(false, JobPushResult.NO_JOB);
            }
    		//调用方法推送任务
            SendResult sendResult = invoker.invoke(jobPos);
    		
    		//记录日志
            if (sendResult.isSuccess()) {
                List<JobLogPo> jobLogPos = new ArrayList<JobLogPo>(jobPos.size());
                for (JobPo jobPo : jobPos) {
                    // 记录日志
                    JobLogPo jobLogPo = JobDomainConverter.convertJobLog(jobPo);
                    jobLogPo.setSuccess(true);
                    jobLogPo.setLogType(LogType.SENT);
                    jobLogPo.setLogTime(SystemClock.now());
                    jobLogPo.setLevel(Level.INFO);
                    jobLogPos.add(jobLogPo);
                }
                appContext.getJobLogger().log(jobLogPos);
            }
            return sendResult;
        }

总结：

JobTracker主要两方面功能

（1）接收JobClient的任务，将任务进行持久化操作

（2）接收TaskTracker的通知，当存在JobClient的任务时将任务信息推送到TaskTracker中。

[LTS_--JobClient]: https://blog.csdn.net/qq924862077/article/details/82813165