Browse Source

Refactor worker (#2121)

* refactor worker registry

* refactor master server

* refactor MasterSchedulerService
pull/2/head
Tboy 5 years ago committed by GitHub
parent
commit
fd6f13fff7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 7
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/MasterServer.java
  2. 26
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerService.java

7
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/MasterServer.java

@ -26,6 +26,7 @@ import org.apache.dolphinscheduler.server.master.processor.TaskAckProcessor;
import org.apache.dolphinscheduler.server.master.processor.TaskKillResponseProcessor; import org.apache.dolphinscheduler.server.master.processor.TaskKillResponseProcessor;
import org.apache.dolphinscheduler.server.master.processor.TaskResponseProcessor; import org.apache.dolphinscheduler.server.master.processor.TaskResponseProcessor;
import org.apache.dolphinscheduler.server.master.registry.MasterRegistry; import org.apache.dolphinscheduler.server.master.registry.MasterRegistry;
import org.apache.dolphinscheduler.server.master.runner.MasterSchedulerService;
import org.apache.dolphinscheduler.server.zk.ZKMasterClient; import org.apache.dolphinscheduler.server.zk.ZKMasterClient;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext; import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.quartz.QuartzExecutors; import org.apache.dolphinscheduler.service.quartz.QuartzExecutors;
@ -80,6 +81,9 @@ public class MasterServer {
@Autowired @Autowired
private ZKMasterClient zkMasterClient; private ZKMasterClient zkMasterClient;
@Autowired
private MasterSchedulerService masterSchedulerService;
/** /**
* master server startup * master server startup
* *
@ -109,6 +113,8 @@ public class MasterServer {
// //
this.zkMasterClient.start(); this.zkMasterClient.start();
this.masterRegistry.registry(); this.masterRegistry.registry();
//
masterSchedulerService.start();
// start QuartzExecutors // start QuartzExecutors
// what system should do if exception // what system should do if exception
@ -162,6 +168,7 @@ public class MasterServer {
this.nettyRemotingServer.close(); this.nettyRemotingServer.close();
this.masterRegistry.unRegistry(); this.masterRegistry.unRegistry();
this.zkMasterClient.close(); this.zkMasterClient.close();
this.masterSchedulerService.close();
//close quartz //close quartz
try{ try{

26
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerService.java

@ -34,8 +34,6 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.ThreadPoolExecutor;
/** /**
@ -49,11 +47,6 @@ public class MasterSchedulerService extends Thread {
*/ */
private static final Logger logger = LoggerFactory.getLogger(MasterSchedulerService.class); private static final Logger logger = LoggerFactory.getLogger(MasterSchedulerService.class);
/**
* master exec service
*/
private ThreadPoolExecutor masterExecService;
/** /**
* dolphinscheduler database interface * dolphinscheduler database interface
*/ */
@ -66,28 +59,33 @@ public class MasterSchedulerService extends Thread {
@Autowired @Autowired
private ZKMasterClient zkMasterClient; private ZKMasterClient zkMasterClient;
@Autowired
private MasterConfig masterConfig;
/** /**
* netty remoting client * netty remoting client
*/ */
private NettyRemotingClient nettyRemotingClient; private final NettyRemotingClient nettyRemotingClient;
@Autowired /**
private MasterConfig masterConfig; * master exec service
*/
private final ThreadPoolExecutor masterExecService;
/** /**
* constructor of MasterSchedulerThread * constructor of MasterSchedulerThread
*/ */
@PostConstruct public MasterSchedulerService(){
public void init(){
this.masterExecService = (ThreadPoolExecutor)ThreadUtils.newDaemonFixedThreadExecutor("Master-Exec-Thread", masterConfig.getMasterExecThreads()); this.masterExecService = (ThreadPoolExecutor)ThreadUtils.newDaemonFixedThreadExecutor("Master-Exec-Thread", masterConfig.getMasterExecThreads());
NettyClientConfig clientConfig = new NettyClientConfig(); NettyClientConfig clientConfig = new NettyClientConfig();
this.nettyRemotingClient = new NettyRemotingClient(clientConfig); this.nettyRemotingClient = new NettyRemotingClient(clientConfig);
}
public void start(){
super.setName("MasterSchedulerThread"); super.setName("MasterSchedulerThread");
super.start(); super.start();
} }
@PreDestroy
public void close(){ public void close(){
nettyRemotingClient.close(); nettyRemotingClient.close();
logger.info("master schedule service stopped..."); logger.info("master schedule service stopped...");

Loading…
Cancel
Save