（二）3.1.9 生产“稳”担当：Apache DolphinScheduler Worker 服务源码全方位解析
作者 | 李杰 移动云，Apache DolphinScheduler贡献者在现代数据驱动的企业中，工作流调度系统是数据管道（Data Pipeline）的“中枢神经”。从 ETL 任务到机器学习训练，从报表生成到实时监控，几乎所有关键业务都依赖于一个稳定、高效、易扩展的调度引擎。
笔者认为 Apache DolphinScheduler 3.1.9 是稳定且广泛使用的版本，故本系列文章将深入其源码核心，剖析其架构设计、模块划分与关键实现机制，帮助开发者理解 Master 和 Worker “如何工作”，并为进一步二次开发或性能优化打下基础。
我们之前解读了 Apache DolphinScheduler 3.1.9版本源码的 Master server 启动流程，感兴趣的可以去查看。本文是 Apache DolphinScheduler 3.1.9 版本源码解读的第二篇：Worker Server 启动流程源码解读以及相关流程设计。结尾处附有相关流程图，供大家参考。
2. Worker Server启动核心概览
[*]代码入口：org.apache.dolphinscheduler.server.worker.WorkerServer#run
public void run() {
// 1. rpc启动
this.workerRpcServer.start();
// 忽略，因为workerRpcServer初始化时包含workerRpcClient初始化的功能
this.workerRpcClient.start();
// 2. 任务插件初始化
this.taskPluginManager.loadPlugin();
this.workerRegistryClient.setRegistryStoppable(this);
// 3. worker 注册
this.workerRegistryClient.start();
// 4. worker管理线程，不断从任务队列中waitSubmitQueue领取任务，提交到线程池处理
this.workerManagerThread.start();
// 5. 消息重试线程。负责轮询通过RPC发送服务，如当task在运行中，若未收到master的ack信息，会周期给master发送“运行中”信号
this.messageRetryRunner.start();
...
}2.1 rpc启动:
[*]描述：注册相关命令的process处理器，如接收任务请求、停止任务请求等。
[*]代码入口：org.apache.dolphinscheduler.server.worker.rpc.WorkerRpcServer#start
public void start() {
LOGGER.info("Worker rpc server starting");
NettyServerConfig serverConfig = new NettyServerConfig();
serverConfig.setListenPort(workerConfig.getListenPort());
this.nettyRemotingServer = new NettyRemotingServer(serverConfig);
// 接收派发任务请求。然后将任务放置到任务队列waitSubmitQueue中，等待workerManagerThread去处理
this.nettyRemotingServer.registerProcessor(CommandType.TASK_DISPATCH_REQUEST, taskDispatchProcessor);
// 停止任务请求
this.nettyRemotingServer.registerProcessor(CommandType.TASK_KILL_REQUEST, taskKillProcessor);
// 接收任务运行中的ack请求
this.nettyRemotingServer.registerProcessor(CommandType.TASK_EXECUTE_RUNNING_ACK,
taskExecuteRunningAckProcessor);
// 接收任务结果的ack请求
this.nettyRemotingServer.registerProcessor(CommandType.TASK_EXECUTE_RESULT_ACK, taskExecuteResultAckProcessor);
this.nettyRemotingServer.registerProcessor(CommandType.TASK_REJECT_ACK, taskRejectAckProcessor);
this.nettyRemotingServer.registerProcessor(CommandType.PROCESS_HOST_UPDATE_REQUEST, hostUpdateProcessor);
this.nettyRemotingServer.registerProcessor(CommandType.TASK_SAVEPOINT_REQUEST, taskSavePointProcessor);
// logger server
this.nettyRemotingServer.registerProcessor(CommandType.GET_APP_ID_REQUEST, loggerRequestProcessor);
this.nettyRemotingServer.registerProcessor(CommandType.GET_LOG_BYTES_REQUEST, loggerRequestProcessor);
this.nettyRemotingServer.registerProcessor(CommandType.ROLL_VIEW_LOG_REQUEST, loggerRequestProcessor);
this.nettyRemotingServer.registerProcessor(CommandType.VIEW_WHOLE_LOG_REQUEST, loggerRequestProcessor);
this.nettyRemotingServer.registerProcessor(CommandType.REMOVE_TAK_LOG_REQUEST, loggerRequestProcessor);
this.nettyRemotingServer.start();
LOGGER.info("Worker rpc server started");
}此处以TASK_DISPATCH_REQUEST为例进行描述。当有任务从master派发请求时，worker会接受TASK_DISPATCH_REQUEST的RPC请求，然后触发process处理器taskDispatchProcessor(org.apache.dolphinscheduler.server.worker.processor.TaskDispatchProcessor#process)的处理:
public void process(Channel channel, Command command) {
...
TaskExecutionContext taskExecutionContext = taskDispatchCommand.getTaskExecutionContext();
...
// set cache, it will be used when kill task
TaskExecutionContextCacheManager.cacheTaskExecutionContext(taskExecutionContext);
// 设置执行任务的worker地址
taskExecutionContext.setHost(workerConfig.getWorkerAddress());
// 设置任务执行日志的目录
taskExecutionContext.setLogPath(LogUtils.getTaskLogPath(taskExecutionContext));
// 构建任务执行线程。整个任务执行需要依赖该线程
WorkerDelayTaskExecuteRunnable workerTaskExecuteRunnable = WorkerTaskExecuteRunnableFactoryBuilder
.createWorkerDelayTaskExecuteRunnableFactory(
taskExecutionContext,
workerConfig,
workflowMasterAddress,
workerMessageSender,
alertClientService,
taskPluginManager,
storageOperate)
.createWorkerTaskExecuteRunnable();
// submit task to manager
// 提交到一个task队列，然后有消费者消费该队列
boolean offer = workerManager.offer(workerTaskExecuteRunnable);
...
}最终会提交给waitSubmitQueue队列，后续有消费者不断进行消费。
public boolean offer(WorkerDelayTaskExecuteRunnable workerDelayTaskExecuteRunnable) {
if (workerConfig.getTaskExecuteThreadsFullPolicy() == TaskExecuteThreadsFullPolicy.CONTINUE) {
return waitSubmitQueue.offer(workerDelayTaskExecuteRunnable);
}
if (waitSubmitQueue.size() > workerExecThreads) {
logger.warn("Wait submit queue is full, will retry submit task later");
WorkerServerMetrics.incWorkerSubmitQueueIsFullCount();
// if waitSubmitQueue is full, it will wait 1s, then try add
ThreadUtils.sleep(Constants.SLEEP_TIME_MILLIS);
if (waitSubmitQueue.size() > workerExecThreads) {
return false;
}
}
return waitSubmitQueue.offer(workerDelayTaskExecuteRunnable);
}2.2 任务插件初始化：
[*]描述：task的相关模板操作，如创建task、解析task参数、获取task资源信息等。对于该插件，api、master、worker都需要进行注册，在worker的作用是获取文件资源、创建任务信息等。
2.3 worker 注册：
[*]描述：将worker信息注册至注册中心(本文以zookeeper为例)，同时监听注册变化情况。
[*]代码入口：org.apache.dolphinscheduler.server.worker.registry.WorkerRegistryClient#start
public void start() {
try {
// 1、将worker信息注册至注册中心(本文以zookeeper为例)
registry();
// 2、监听自身与注册中心的连接情况；
registryClient.addConnectionStateListener(
new WorkerConnectionStateListener(workerConfig, registryClient, workerConnectStrategy));
} catch (Exception ex) {
throw new RegistryException("Worker registry client start up error", ex);
}
}2.4 worker管理线程：
[*]描述：不断从任务队列中waitSubmitQueue领取任务，提交到线程池处理。
[*]代码入口：org.apache.dolphinscheduler.server.worker.runner.WorkerManagerThread#run
public void run() { Thread.currentThread().setName("Worker-Execute-Manager-Thread"); while (!ServerLifeCycleManager.isStopped()) { try { if (!ServerLifeCycleManager.isRunning()) { Thread.sleep(Constants.SLEEP_TIME_MILLIS); } // 1、如果任务线程池线程个数够用，则处理任务 if (this.getThreadPoolQueueSize() 感谢分享，学习下。
页:
[1]