|
|
|
|
@ -17,22 +17,14 @@
|
|
|
|
|
package org.apache.dolphinscheduler.common.queue; |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
import org.apache.commons.configuration.Configuration; |
|
|
|
|
import org.apache.commons.configuration.ConfigurationException; |
|
|
|
|
import org.apache.commons.configuration.PropertiesConfiguration; |
|
|
|
|
import org.apache.curator.framework.CuratorFramework; |
|
|
|
|
import org.apache.curator.framework.CuratorFrameworkFactory; |
|
|
|
|
import org.apache.curator.retry.ExponentialBackoffRetry; |
|
|
|
|
import org.apache.dolphinscheduler.common.Constants; |
|
|
|
|
import org.apache.dolphinscheduler.common.utils.Bytes; |
|
|
|
|
import org.apache.dolphinscheduler.common.utils.IpUtils; |
|
|
|
|
import org.apache.dolphinscheduler.common.utils.OSUtils; |
|
|
|
|
import org.apache.dolphinscheduler.common.zk.DefaultEnsembleProvider; |
|
|
|
|
import org.apache.dolphinscheduler.common.zk.ZookeeperConfig; |
|
|
|
|
import org.apache.zookeeper.CreateMode; |
|
|
|
|
import org.apache.zookeeper.data.Stat; |
|
|
|
|
import org.apache.dolphinscheduler.common.zk.ZookeeperOperator; |
|
|
|
|
import org.slf4j.Logger; |
|
|
|
|
import org.slf4j.LoggerFactory; |
|
|
|
|
import org.springframework.beans.factory.annotation.Autowired; |
|
|
|
|
import org.springframework.stereotype.Service; |
|
|
|
|
|
|
|
|
|
import java.util.*; |
|
|
|
|
|
|
|
|
|
@ -40,35 +32,13 @@ import java.util.*;
|
|
|
|
|
* A singleton of a task queue implemented with zookeeper |
|
|
|
|
* tasks queue implemention |
|
|
|
|
*/ |
|
|
|
|
@Service |
|
|
|
|
public class TaskQueueZkImpl implements ITaskQueue { |
|
|
|
|
|
|
|
|
|
private static final Logger logger = LoggerFactory.getLogger(TaskQueueZkImpl.class); |
|
|
|
|
|
|
|
|
|
private static volatile TaskQueueZkImpl instance; |
|
|
|
|
|
|
|
|
|
private CuratorFramework zkClient; |
|
|
|
|
|
|
|
|
|
private ZookeeperConfig zookeeperConfig; |
|
|
|
|
|
|
|
|
|
private CuratorFramework getZkClient() { |
|
|
|
|
return zkClient; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private TaskQueueZkImpl(){ |
|
|
|
|
init(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
public static TaskQueueZkImpl getInstance(){ |
|
|
|
|
if (null == instance) { |
|
|
|
|
synchronized (TaskQueueZkImpl.class) { |
|
|
|
|
if(null == instance) { |
|
|
|
|
instance = new TaskQueueZkImpl(); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
return instance; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@Autowired |
|
|
|
|
private ZookeeperOperator zookeeperOperator; |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* get all tasks from tasks queue |
|
|
|
|
@ -78,14 +48,12 @@ public class TaskQueueZkImpl implements ITaskQueue {
|
|
|
|
|
@Override |
|
|
|
|
public List<String> getAllTasks(String key) { |
|
|
|
|
try { |
|
|
|
|
List<String> list = getZkClient().getChildren().forPath(getTasksPath(key)); |
|
|
|
|
|
|
|
|
|
List<String> list = zookeeperOperator.getChildrenKeys(getTasksPath(key)); |
|
|
|
|
return list; |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
logger.error("get all tasks from tasks queue exception",e); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
return new ArrayList<String>(); |
|
|
|
|
return new ArrayList<>(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
@ -99,22 +67,8 @@ public class TaskQueueZkImpl implements ITaskQueue {
|
|
|
|
|
public boolean checkTaskExists(String key, String task) { |
|
|
|
|
String taskPath = getTasksPath(key) + Constants.SINGLE_SLASH + task; |
|
|
|
|
|
|
|
|
|
try { |
|
|
|
|
Stat stat = zkClient.checkExists().forPath(taskPath); |
|
|
|
|
|
|
|
|
|
if(null == stat){ |
|
|
|
|
logger.info("check task:{} not exist in task queue",task); |
|
|
|
|
return false; |
|
|
|
|
}else{ |
|
|
|
|
logger.info("check task {} exists in task queue ",task); |
|
|
|
|
return true; |
|
|
|
|
} |
|
|
|
|
return zookeeperOperator.isExisted(taskPath); |
|
|
|
|
|
|
|
|
|
} catch (Exception e) { |
|
|
|
|
logger.info(String.format("task {} check exists in task queue exception ", task), e); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
return false; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@ -128,9 +82,7 @@ public class TaskQueueZkImpl implements ITaskQueue {
|
|
|
|
|
public boolean add(String key, String value){ |
|
|
|
|
try { |
|
|
|
|
String taskIdPath = getTasksPath(key) + Constants.SINGLE_SLASH + value; |
|
|
|
|
String result = getZkClient().create().withMode(CreateMode.PERSISTENT).forPath(taskIdPath, Bytes.toBytes(value)); |
|
|
|
|
|
|
|
|
|
logger.info("add task : {} to tasks queue , result success",result); |
|
|
|
|
zookeeperOperator.persist(taskIdPath, value); |
|
|
|
|
return true; |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
logger.error("add task to tasks queue exception",e); |
|
|
|
|
@ -153,8 +105,7 @@ public class TaskQueueZkImpl implements ITaskQueue {
|
|
|
|
|
@Override |
|
|
|
|
public List<String> poll(String key, int tasksNum) { |
|
|
|
|
try{ |
|
|
|
|
CuratorFramework zk = getZkClient(); |
|
|
|
|
List<String> list = zk.getChildren().forPath(getTasksPath(key)); |
|
|
|
|
List<String> list = zookeeperOperator.getChildrenKeys(getTasksPath(key)); |
|
|
|
|
|
|
|
|
|
if(list != null && list.size() > 0){ |
|
|
|
|
|
|
|
|
|
@ -277,15 +228,12 @@ public class TaskQueueZkImpl implements ITaskQueue {
|
|
|
|
|
@Override |
|
|
|
|
public void removeNode(String key, String nodeValue){ |
|
|
|
|
|
|
|
|
|
CuratorFramework zk = getZkClient(); |
|
|
|
|
String tasksQueuePath = getTasksPath(key) + Constants.SINGLE_SLASH; |
|
|
|
|
String taskIdPath = tasksQueuePath + nodeValue; |
|
|
|
|
logger.info("consume task {}", taskIdPath); |
|
|
|
|
logger.info("removeNode task {}", taskIdPath); |
|
|
|
|
try{ |
|
|
|
|
Stat stat = zk.checkExists().forPath(taskIdPath); |
|
|
|
|
if(stat != null){ |
|
|
|
|
zk.delete().forPath(taskIdPath); |
|
|
|
|
} |
|
|
|
|
zookeeperOperator.remove(taskIdPath); |
|
|
|
|
|
|
|
|
|
}catch(Exception e){ |
|
|
|
|
logger.error(String.format("delete task:%s from zookeeper fail, exception:" ,nodeValue) ,e); |
|
|
|
|
} |
|
|
|
|
@ -307,13 +255,10 @@ public class TaskQueueZkImpl implements ITaskQueue {
|
|
|
|
|
|
|
|
|
|
if(value != null && value.trim().length() > 0){ |
|
|
|
|
String path = getTasksPath(key) + Constants.SINGLE_SLASH; |
|
|
|
|
CuratorFramework zk = getZkClient(); |
|
|
|
|
Stat stat = zk.checkExists().forPath(path + value); |
|
|
|
|
|
|
|
|
|
if(null == stat){ |
|
|
|
|
String result = zk.create().withMode(CreateMode.PERSISTENT).forPath(path + value,Bytes.toBytes(value)); |
|
|
|
|
logger.info("add task:{} to tasks set result:{} ",value,result); |
|
|
|
|
}else{ |
|
|
|
|
if(!zookeeperOperator.isExisted(path + value)){ |
|
|
|
|
zookeeperOperator.persist(path + value,value); |
|
|
|
|
logger.info("add task:{} to tasks set ",value); |
|
|
|
|
} else{ |
|
|
|
|
logger.info("task {} exists in tasks set ",value); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -336,15 +281,7 @@ public class TaskQueueZkImpl implements ITaskQueue {
|
|
|
|
|
public void srem(String key, String value) { |
|
|
|
|
try{ |
|
|
|
|
String path = getTasksPath(key) + Constants.SINGLE_SLASH; |
|
|
|
|
CuratorFramework zk = getZkClient(); |
|
|
|
|
Stat stat = zk.checkExists().forPath(path + value); |
|
|
|
|
|
|
|
|
|
if(null != stat){ |
|
|
|
|
zk.delete().forPath(path + value); |
|
|
|
|
logger.info("delete task:{} from tasks set ",value); |
|
|
|
|
}else{ |
|
|
|
|
logger.info("delete task:{} from tasks set fail, there is no this task",value); |
|
|
|
|
} |
|
|
|
|
zookeeperOperator.remove(path + value); |
|
|
|
|
|
|
|
|
|
}catch(Exception e){ |
|
|
|
|
logger.error(String.format("delete task:" + value + " exception"),e); |
|
|
|
|
@ -363,7 +300,7 @@ public class TaskQueueZkImpl implements ITaskQueue {
|
|
|
|
|
Set<String> tasksSet = new HashSet<>(); |
|
|
|
|
|
|
|
|
|
try { |
|
|
|
|
List<String> list = getZkClient().getChildren().forPath(getTasksPath(key)); |
|
|
|
|
List<String> list = zookeeperOperator.getChildrenKeys(getTasksPath(key)); |
|
|
|
|
|
|
|
|
|
for (String task : list) { |
|
|
|
|
tasksSet.add(task); |
|
|
|
|
@ -377,56 +314,6 @@ public class TaskQueueZkImpl implements ITaskQueue {
|
|
|
|
|
return tasksSet; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* Init the task queue of zookeeper node |
|
|
|
|
*/ |
|
|
|
|
private void init(){ |
|
|
|
|
initZkClient(); |
|
|
|
|
try { |
|
|
|
|
String tasksQueuePath = getTasksPath(Constants.DOLPHINSCHEDULER_TASKS_QUEUE); |
|
|
|
|
String tasksCancelPath = getTasksPath(Constants.DOLPHINSCHEDULER_TASKS_KILL); |
|
|
|
|
|
|
|
|
|
for(String taskQueuePath : new String[]{tasksQueuePath,tasksCancelPath}){ |
|
|
|
|
if(zkClient.checkExists().forPath(taskQueuePath) == null){ |
|
|
|
|
// create a persistent parent node
|
|
|
|
|
zkClient.create().creatingParentContainersIfNeeded() |
|
|
|
|
.withMode(CreateMode.PERSISTENT).forPath(taskQueuePath); |
|
|
|
|
logger.info("create tasks queue parent node success : {} ",taskQueuePath); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
} catch (Exception e) { |
|
|
|
|
logger.error("create zk node failure",e); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private void initZkClient() { |
|
|
|
|
|
|
|
|
|
Configuration conf = null; |
|
|
|
|
try { |
|
|
|
|
conf = new PropertiesConfiguration(Constants.ZOOKEEPER_PROPERTIES_PATH); |
|
|
|
|
} catch (ConfigurationException ex) { |
|
|
|
|
logger.error("load zookeeper properties file failed, system exit"); |
|
|
|
|
System.exit(-1); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
zkClient = CuratorFrameworkFactory.builder().ensembleProvider(new DefaultEnsembleProvider(conf.getString("zookeeper.quorum"))) |
|
|
|
|
.retryPolicy(new ExponentialBackoffRetry(conf.getInt("zookeeper.retry.base.sleep"), conf.getInt("zookeeper.retry.maxtime"), conf.getInt("zookeeper.retry.max.sleep"))) |
|
|
|
|
.sessionTimeoutMs(conf.getInt("zookeeper.session.timeout")) |
|
|
|
|
.connectionTimeoutMs(conf.getInt("zookeeper.connection.timeout")) |
|
|
|
|
.build(); |
|
|
|
|
|
|
|
|
|
zkClient.start(); |
|
|
|
|
try { |
|
|
|
|
zkClient.blockUntilConnected(); |
|
|
|
|
} catch (final Exception ex) { |
|
|
|
|
throw new RuntimeException(ex); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* Clear the task queue of zookeeper node |
|
|
|
|
*/ |
|
|
|
|
@ -437,16 +324,12 @@ public class TaskQueueZkImpl implements ITaskQueue {
|
|
|
|
|
String tasksCancelPath = getTasksPath(Constants.DOLPHINSCHEDULER_TASKS_KILL); |
|
|
|
|
|
|
|
|
|
for(String taskQueuePath : new String[]{tasksQueuePath,tasksCancelPath}){ |
|
|
|
|
if(zkClient.checkExists().forPath(taskQueuePath) != null){ |
|
|
|
|
|
|
|
|
|
List<String> list = zkClient.getChildren().forPath(taskQueuePath); |
|
|
|
|
|
|
|
|
|
if(zookeeperOperator.isExisted(taskQueuePath)){ |
|
|
|
|
List<String> list = zookeeperOperator.getChildrenKeys(taskQueuePath); |
|
|
|
|
for (String task : list) { |
|
|
|
|
zkClient.delete().forPath(taskQueuePath + Constants.SINGLE_SLASH + task); |
|
|
|
|
zookeeperOperator.remove(taskQueuePath + Constants.SINGLE_SLASH + task); |
|
|
|
|
logger.info("delete task from tasks queue : {}/{} ",taskQueuePath,task); |
|
|
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
dk.technoboy
5 years ago