Apache DolphinScheduler-1.3.9源码分析（二）

引言

随着大数据的发展，任务调度系统成为了数据处理和管理中至关重要的部分。Apache DolphinScheduler 是一款优秀的开源分布式工作流调度平台，在大数据场景中得到广泛应用。

在本文中，我们将对 Apache DolphinScheduler 1.3.9 版本的源码进行深入分析，主要分析一下Master和Worker的交互设计。
感兴趣的朋友也可以回顾我们上一篇文章：Apache DolphinScheduler-1.3.9源码分析（一）
Worker配置文件

1. # worker listener port
2. worker.listen.port=1234
4. # worker execute thread number to limit task instances in parallel
5. # worker可并行的任务数限制
6. worker.exec.threads=100
8. # worker heartbeat interval, the unit is second
9. # worker发送心跳间隔
10. worker.heartbeat.interval=10
12. # worker max cpuload avg, only higher than the system cpu load average, worker server can be dispatched tasks. default value -1: the number of cpu cores * 2
13. # worker最大cpu平均负载，只有系统cpu平均负载低于该值，才能执行任务
14. # 默认值为-1，则最大cpu平均负载=系统cpu核数 * 2
15. worker.max.cpuload.avg=-1
17. # worker reserved memory, only lower than system available memory, worker server can be dispatched tasks. default value 0.3, the unit is G
18. # worker的预留内存，只有当系统可用内存大于等于该值，才能执行任务，单位为GB
19. # 默认0.3G
20. worker.reserved.memory=0.3
22. # default worker groups separated by comma, like 'worker.groups=default,test'
23. # 工作组名称，多个用,隔开
24. worker.groups=default

复制代码

WorkerServer启动

1. public void run() {
2.     // init remoting server
3.     NettyServerConfig serverConfig = new NettyServerConfig();
4.     serverConfig.setListenPort(workerConfig.getListenPort());
5.     this.nettyRemotingServer = new NettyRemotingServer(serverConfig);
6.     this.nettyRemotingServer.registerProcessor(CommandType.TASK_EXECUTE_REQUEST, new TaskExecuteProcessor());
7.     this.nettyRemotingServer.registerProcessor(CommandType.TASK_KILL_REQUEST, new TaskKillProcessor());
8.     this.nettyRemotingServer.registerProcessor(CommandType.DB_TASK_ACK, new DBTaskAckProcessor());
9.     this.nettyRemotingServer.registerProcessor(CommandType.DB_TASK_RESPONSE, new DBTaskResponseProcessor());
10.     this.nettyRemotingServer.start();
12.     // worker registry
13.     try {
14.         this.workerRegistry.registry();
15.         this.workerRegistry.getZookeeperRegistryCenter().setStoppable(this);
16.         Set<String> workerZkPaths = this.workerRegistry.getWorkerZkPaths();
17.         this.workerRegistry.getZookeeperRegistryCenter().getRegisterOperator().handleDeadServer(workerZkPaths, ZKNodeType.WORKER, Constants.DELETE_ZK_OP);
18.     } catch (Exception e) {
19.         logger.error(e.getMessage(), e);
20.         throw new RuntimeException(e);
21.     }
23.     // retry report task status
24.     this.retryReportTaskStatusThread.start();
26.     /**
27.      * register hooks, which are called before the process exits
28.      */
29.     Runtime.getRuntime().addShutdownHook(new Thread(() -> {
30.         if (Stopper.isRunning()) {
31.             close("shutdownHook");
32.         }
33.     }));
34. }

复制代码

注册四个Command：

• TASK_EXECUTE_REQUEST：task执行请求
  
• TASK_KILL_REQUEST：task停止请求
  
• DB_TASK_ACK：Worker接受到Master的调度请求，回应master
  
• DB_TASK_RESPONSE：
  

• 注册WorkerServer到Zookeeper，并发送心跳
  
• 报告Task执行状态
  

RetryReportTaskStatusThread

这是一个兜底机制，主要负责定时轮询向Master汇报任务的状态，直到Master回复状态的ACK，避免任务状态丢失；
每隔5分钟，检查一下responceCache中的ACK Cache和Response Cache是否为空，如果不为空则向Master发送ack_command和response command请求。

1. public void run() {
2.     ResponceCache responceCache = ResponceCache.get();
4.     while (Stopper.isRunning()){
6.         // sleep 5 minutes
7.         ThreadUtils.sleep(RETRY_REPORT_TASK_STATUS_INTERVAL);
9.         try {
10.             if (!responceCache.getAckCache().isEmpty()){
11.                 Map<Integer,Command> ackCache =  responceCache.getAckCache();
12.                 for (Map.Entry<Integer, Command> entry : ackCache.entrySet()){
13.                     Integer taskInstanceId = entry.getKey();
14.                     Command ackCommand = entry.getValue();
15.                     taskCallbackService.sendAck(taskInstanceId,ackCommand);
16.                 }
17.             }
19.             if (!responceCache.getResponseCache().isEmpty()){
20.                 Map<Integer,Command> responseCache =  responceCache.getResponseCache();
21.                 for (Map.Entry<Integer, Command> entry : responseCache.entrySet()){
22.                     Integer taskInstanceId = entry.getKey();
23.                     Command responseCommand = entry.getValue();
24.                     taskCallbackService.sendResult(taskInstanceId,responseCommand);
25.                 }
26.             }
27.         }catch (Exception e){
28.             logger.warn("retry report task status error", e);
29.         }
30.     }
31. }

复制代码

Master与Worker的交互设计

Apache DolphinScheduler Master和Worker模块是两个独立的JVM进程，可以部署在不同的服务器上，Master与Worker的通信都是通过Netty实现RPC交互的，一共用到7种处理器。
模块处理器作用mastermasterTaskResponseProcessor处理TaskExecuteResponseCommand消息，将消息添加到TaskResponseService的任务响应队列中mastermasterTaskAckProcessor处理TaskExecuteAckCommand消息，将消息添加到TaskResponseService的任务响应队列中mastermasterTaskKillResponseProcessor处理TaskKillResponseCommand消息，并在日志中打印消息内容workerworkerTaskExecuteProcessor处理TaskExecuteRequestCommand消息，并发送TaskExecuteAckCommand到master，提交任务执行workerworkerTaskKillProcessor处理TaskKillRequestCommand消息，调用kill -9 pid杀死任务对应的进程，并向master发送TaskKillResponseCommand消息workerworkerDBTaskAckProcessor处理DBTaskAckCommand消息，针对执行成功的任务，从ResponseCache中删除workerworkerDBTaskResponseProcessor处理DBTaskResponseCommand消息，针对执行成功的任务，从ResponseCache中删除分发任务如何交互

master#TaskPriorityQueueConsumer

Master任务里有一个TaskPriorityQueueConsumer，会从TaskPriorityQueue里每次取3个Task分发给Worker执行，这里会创建TaskExecuteRequestCommand。
TaskPriorityQueueConsumer#run()

[code]@Overridepublic void run() {    List failedDispatchTasks = new ArrayList();    while (Stopper.isRunning()){        try {            // 每一批次分发任务数量，master.dispatch.task.num = 3            int fetchTaskNum = masterConfig.getMasterDispatchTaskNumber();            failedDispatchTasks.clear();            for(int i = 0; i < fetchTaskNum; i++){                if(taskPriorityQueue.size()