LTS原理--JobClient提交任务过程（二）

朴灿烈づ我的快乐病毒、 2022-05-11 10:26 187阅读 0赞

JobClient是任务的发起者，JobClient的作用就是将相关的任务信息提交到JobTracker中，然后等待接收JobTracker的任务执行结果。

### 一、示例： ###

官方示例地址：[https://github.com/ltsopensource/lts-examples][https_github.com_ltsopensource_lts-examples]

1、属性配置

配置zk的注册地址，其中JobTracker也是注册到zk中，因此JobClient可以从zk中获取JobTracker的执行地址

lts.jobclient.cluster-name=test_cluster
    lts.jobclient.registry-address=zookeeper://127.0.0.1:2181
    lts.jobclient.node-group=test_jobClient
    lts.jobclient.use-retry-client=true
    lts.jobclient.configs.job.fail.store=mapdb

2、注入JobClient

3、提交任务

public class Main {
    
        public static void main(String[] args) {
            ApplicationContext context = new ClassPathXmlApplicationContext("/lts-jobclient.xml");
    
            JobClient jobClient = (JobClient) context.getBean("jobClient");
    
            Job job = new Job();
            job.setTaskId("t_realtime_5556666");
            job.setParam("shopId", "1122222221");
            job.setTaskTrackerNodeGroup("test_trade_TaskTracker");
            job.setNeedFeedback(true);
            job.setReplaceOnExist(true);        // 当任务队列中存在这个任务的时候，是否替换更新
            Response response = jobClient.submitJob(job);
            System.out.println(response);
        }
    
    }

### 二、运行流程 ###

我们通过跟踪代码来查看jobClient提交任务和获取任务执行结果的流程，接下来我们看看任务提交和获取任务执行结果的流程。

Response response = jobClient.submitJob(job)完成了任务提交和运行结果返回操作。

在JobClient中异步线程提交任务

//提交任务
    public Response submitJob(Job job) throws JobSubmitException {
            checkStart();
            return protectSubmit(Collections.singletonList(job));
        }
    
        private Response protectSubmit(List<Job> jobs) throws JobSubmitException {
            //线程池异步提交任务
            return protector.execute(jobs, new JobSubmitExecutor<Response>() {
                @Override
                public Response execute(List<Job> jobs) throws JobSubmitException {
                    return submitJob(jobs, SubmitType.ASYNC);
                }
            });
        }

在submitJob中根据任务创建任务提交请求，创建任务回调方法，根据提交方法选择asyncSubmit和syncSubmit提交任务

protected Response submitJob(final List<Job> jobs, SubmitType type) throws JobSubmitException {
            // 检查参数
            checkFields(jobs);
    
            final Response response = new Response();
            try {
    			//创建任务提交请求
                JobSubmitRequest jobSubmitRequest = CommandBodyWrapper.wrapper(appContext, new JobSubmitRequest());
                jobSubmitRequest.setJobs(jobs);
    
                RemotingCommand requestCommand = RemotingCommand.createRequestCommand(
                        JobProtos.RequestCode.SUBMIT_JOB.code(), jobSubmitRequest);
    
    			//创建任务提交及回调方法
                SubmitCallback submitCallback = new SubmitCallback() {
                    @Override
                    public void call(RemotingCommand responseCommand) {
                        if (responseCommand == null) {
                            response.setFailedJobs(jobs);
                            response.setSuccess(false);
                            response.setMsg("Submit Job failed: JobTracker is broken");
                            LOGGER.warn("Submit Job failed: {}, {}", jobs, "JobTracker is broken");
                            return;
                        }
    
                        if (JobProtos.ResponseCode.JOB_RECEIVE_SUCCESS.code() == responseCommand.getCode()) {
                            if (LOGGER.isDebugEnabled()) {
                                LOGGER.debug("Submit Job success: {}", jobs);
                            }
                            response.setSuccess(true);
                            return;
                        }
                        // 失败的job
                        JobSubmitResponse jobSubmitResponse = responseCommand.getBody();
                        response.setFailedJobs(jobSubmitResponse.getFailedJobs());
                        response.setSuccess(false);
                        response.setCode(JobProtos.ResponseCode.valueOf(responseCommand.getCode()).name());
                        response.setMsg("Submit Job failed: " + responseCommand.getRemark() + " " + jobSubmitResponse.getMsg());
                        LOGGER.warn("Submit Job failed: {}, {}, {}", jobs, responseCommand.getRemark(), jobSubmitResponse.getMsg());
                    }
                };
    			//根据任务提交方式提交任务
                if (SubmitType.ASYNC.equals(type)) {
                    asyncSubmit(requestCommand, submitCallback);
                } else {
                    syncSubmit(requestCommand, submitCallback);
                }
            } catch (JobTrackerNotFoundException e) {
                response.setSuccess(false);
                response.setCode(ResponseCode.JOB_TRACKER_NOT_FOUND);
                response.setMsg("Can not found JobTracker node!");
            } catch (Exception e) {
                response.setSuccess(false);
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setMsg(StringUtils.toString(e));
            } finally {
                // 统计
                if (response.isSuccess()) {
                    stat.incSubmitSuccessNum(jobs.size());
                } else {
                    stat.incSubmitFailedNum(CollectionUtils.sizeOf(response.getFailedJobs()));
                }
            }
    
            return response;
        }

在syncSubmit或asyncSubmit中调用remoteingClient的inokeSync方法提交任务

private void syncSubmit(RemotingCommand requestCommand, final SubmitCallback submitCallback)
                throws JobTrackerNotFoundException {
            submitCallback.call(remotingClient.invokeSync(requestCommand));
        }

在invokeSync方法中会从注册中心中获取JobTracker节点的节点信息，并且会根据一些策略选择（一致性hash，随机、轮训和主从策略）

各种类型节点zk中的信息

![70][]

public RemotingCommand invokeSync(RemotingCommand request)
                throws JobTrackerNotFoundException {
    
    		//从注册中心中获取JobTracker节点的节点信息，会有一些策略
            Node jobTracker = getJobTrackerNode();
    
            try {
    			//rpc远程调用
                RemotingCommand response = remotingClient.invokeSync(jobTracker.getAddress(),
                        request, appContext.getConfig().getInvokeTimeoutMillis());
                this.serverEnable = true;
    			//返回调用结果
                return response;
            } catch (Exception e) {
                // 将这个JobTracker移除
                jobTrackers.remove(jobTracker);
                try {
                    Thread.sleep(100L);
                } catch (InterruptedException e1) {
                    LOGGER.error(e1.getMessage(), e1);
                }
                // 只要不是节点 不可用, 轮询所有节点请求
                return invokeSync(request);
            }
        }

在invokeSync中封装了一些通信方式（netty，mina和lts）与JobTracker通信，并返回执行结果。

@Override
        public RemotingCommand invokeSync(String addr, final RemotingCommand request, long timeoutMillis)
                throws InterruptedException, RemotingConnectException, RemotingSendRequestException,
                RemotingTimeoutException {
            final Channel channel = this.getAndCreateChannel(addr);
            if (channel != null && channel.isConnected()) {
                try {
                    //远程通信，底层Netty、Mina或LTS实现
                    return this.invokeSyncImpl(channel, request, timeoutMillis);
                } catch (RemotingSendRequestException e) {
                    LOGGER.warn("invokeSync: send request exception, so close the channel[{}]", addr);
                    this.closeChannel(addr, channel);
                    throw e;
                } catch (RemotingTimeoutException e) {
                    LOGGER.warn("invokeSync: wait response timeout exception, the channel[{}]", addr);
                    // 超时异常如果关闭连接可能会产生连锁反应
                    // this.closeChannel(addr, channel);
                    throw e;
                }
            } else {
                this.closeChannel(addr, channel);
                throw new RemotingConnectException(addr);
            }
        }

[https_github.com_ltsopensource_lts-examples]: https://github.com/ltsopensource/lts-examples
[70]: /images/20220511/6b66db571dba45e0b30fedb60a7d0d20.png