|
|
|
|
@ -19,6 +19,8 @@ package cn.escheduler.common.queue;
|
|
|
|
|
|
|
|
|
|
import cn.escheduler.common.Constants; |
|
|
|
|
import cn.escheduler.common.utils.Bytes; |
|
|
|
|
import cn.escheduler.common.utils.IpUtils; |
|
|
|
|
import cn.escheduler.common.utils.OSUtils; |
|
|
|
|
import cn.escheduler.common.zk.AbstractZKClient; |
|
|
|
|
import org.apache.curator.framework.CuratorFramework; |
|
|
|
|
import org.apache.zookeeper.CreateMode; |
|
|
|
|
@ -26,10 +28,7 @@ import org.apache.zookeeper.data.Stat;
|
|
|
|
|
import org.slf4j.Logger; |
|
|
|
|
import org.slf4j.LoggerFactory; |
|
|
|
|
|
|
|
|
|
import java.util.ArrayList; |
|
|
|
|
import java.util.HashSet; |
|
|
|
|
import java.util.List; |
|
|
|
|
import java.util.Set; |
|
|
|
|
import java.util.*; |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* A singleton of a task queue implemented with zookeeper |
|
|
|
|
@ -62,7 +61,6 @@ public class TaskQueueZkImpl extends AbstractZKClient implements ITaskQueue {
|
|
|
|
|
* @param key task queue name |
|
|
|
|
* @return |
|
|
|
|
*/ |
|
|
|
|
@Deprecated |
|
|
|
|
@Override |
|
|
|
|
public List<String> getAllTasks(String key) { |
|
|
|
|
try { |
|
|
|
|
@ -80,7 +78,7 @@ public class TaskQueueZkImpl extends AbstractZKClient implements ITaskQueue {
|
|
|
|
|
* check task exists in the task queue or not |
|
|
|
|
* |
|
|
|
|
* @param key queue name |
|
|
|
|
* @param task ${priority}_${processInstanceId}_${taskId} |
|
|
|
|
* @param task ${processInstancePriority}_${processInstanceId}_${taskInstancePriority}_${taskId} |
|
|
|
|
* @return true if exists in the queue |
|
|
|
|
*/ |
|
|
|
|
@Override |
|
|
|
|
@ -110,7 +108,7 @@ public class TaskQueueZkImpl extends AbstractZKClient implements ITaskQueue {
|
|
|
|
|
* add task to tasks queue |
|
|
|
|
* |
|
|
|
|
* @param key task queue name |
|
|
|
|
* @param value ${priority}_${processInstanceId}_${taskId} |
|
|
|
|
* @param value ${processInstancePriority}_${processInstanceId}_${taskInstancePriority}_${taskId}_host1,host2,... |
|
|
|
|
*/ |
|
|
|
|
@Override |
|
|
|
|
public void add(String key, String value) { |
|
|
|
|
@ -118,9 +116,6 @@ public class TaskQueueZkImpl extends AbstractZKClient implements ITaskQueue {
|
|
|
|
|
String taskIdPath = getTasksPath(key) + Constants.SINGLE_SLASH + value; |
|
|
|
|
String result = getZkClient().create().withMode(CreateMode.PERSISTENT).forPath(taskIdPath, Bytes.toBytes(value)); |
|
|
|
|
|
|
|
|
|
// String path = conf.getString(Constants.ZOOKEEPER_SCHEDULER_ROOT) + Constants.SINGLE_SLASH + Constants.SCHEDULER_TASKS_QUEUE + "_add" + Constants.SINGLE_SLASH + value;
|
|
|
|
|
// getZkClient().create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path,
|
|
|
|
|
// Bytes.toBytes(value));
|
|
|
|
|
logger.info("add task : {} to tasks queue , result success",result); |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
logger.error("add task to tasks queue exception",e); |
|
|
|
|
@ -132,16 +127,16 @@ public class TaskQueueZkImpl extends AbstractZKClient implements ITaskQueue {
|
|
|
|
|
/** |
|
|
|
|
* An element pops out of the queue <p> |
|
|
|
|
* note: |
|
|
|
|
* ${processInstancePriority}_${processInstanceId}_${taskInstancePriority}_${taskId} |
|
|
|
|
* ${processInstancePriority}_${processInstanceId}_${taskInstancePriority}_${taskId}_host1,host2,... |
|
|
|
|
* The tasks with the highest priority are selected by comparing the priorities of the above four levels from high to low. |
|
|
|
|
* |
|
|
|
|
* 流程实例优先级_流程实例id_任务优先级_任务id high <- low |
|
|
|
|
* 流程优先级_流程实例id_任务优先级_任务id_任务执行的机器id1,任务执行的机器id2,... high <- low |
|
|
|
|
* @param key task queue name |
|
|
|
|
* @param remove whether remove the element |
|
|
|
|
* @return the task id to be executed |
|
|
|
|
* @param tasksNum how many elements to poll |
|
|
|
|
* @return the task ids to be executed |
|
|
|
|
*/ |
|
|
|
|
@Override |
|
|
|
|
public String poll(String key, boolean remove) { |
|
|
|
|
public List<String> poll(String key, int tasksNum) { |
|
|
|
|
try{ |
|
|
|
|
CuratorFramework zk = getZkClient(); |
|
|
|
|
String tasksQueuePath = getTasksPath(key) + Constants.SINGLE_SLASH; |
|
|
|
|
@ -149,55 +144,79 @@ public class TaskQueueZkImpl extends AbstractZKClient implements ITaskQueue {
|
|
|
|
|
|
|
|
|
|
if(list != null && list.size() > 0){ |
|
|
|
|
|
|
|
|
|
String workerIp = OSUtils.getHost(); |
|
|
|
|
String workerIpLongStr = String.valueOf(IpUtils.ipToLong(workerIp)); |
|
|
|
|
|
|
|
|
|
int size = list.size(); |
|
|
|
|
|
|
|
|
|
String formatTargetTask = null; |
|
|
|
|
String targetTaskKey = null; |
|
|
|
|
|
|
|
|
|
Set<String> taskTreeSet = new TreeSet<>(); |
|
|
|
|
|
|
|
|
|
for (int i = 0; i < size; i++) { |
|
|
|
|
|
|
|
|
|
String taskDetail = list.get(i); |
|
|
|
|
String[] taskDetailArrs = taskDetail.split(Constants.UNDERLINE); |
|
|
|
|
|
|
|
|
|
if(taskDetailArrs.length == 4){ |
|
|
|
|
//向前版本兼容
|
|
|
|
|
if(taskDetailArrs.length >= 4){ |
|
|
|
|
|
|
|
|
|
//format ${processInstancePriority}_${processInstanceId}_${taskInstancePriority}_${taskId}
|
|
|
|
|
String formatTask = String.format("%s_%010d_%s_%010d", taskDetailArrs[0], Long.parseLong(taskDetailArrs[1]), taskDetailArrs[2], Long.parseLong(taskDetailArrs[3])); |
|
|
|
|
if(i > 0){ |
|
|
|
|
int result = formatTask.compareTo(formatTargetTask); |
|
|
|
|
if(result < 0){ |
|
|
|
|
formatTargetTask = formatTask; |
|
|
|
|
targetTaskKey = taskDetail; |
|
|
|
|
if(taskDetailArrs.length > 4){ |
|
|
|
|
String taskHosts = taskDetailArrs[4]; |
|
|
|
|
|
|
|
|
|
//task can assign to any worker host if equals default ip value of worker server
|
|
|
|
|
if(!taskHosts.equals(Constants.DEFAULT_WORKER_ID)){ |
|
|
|
|
String[] taskHostsArr = taskHosts.split(Constants.COMMA); |
|
|
|
|
|
|
|
|
|
if(!Arrays.asList(taskHostsArr).contains(workerIpLongStr)){ |
|
|
|
|
continue; |
|
|
|
|
} |
|
|
|
|
}else{ |
|
|
|
|
formatTargetTask = formatTask; |
|
|
|
|
targetTaskKey = taskDetail; |
|
|
|
|
} |
|
|
|
|
}else{ |
|
|
|
|
logger.error("task queue poll error, task detail :{} , please check!", taskDetail); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
taskTreeSet.add(formatTask); |
|
|
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
if(formatTargetTask != null){ |
|
|
|
|
String taskIdPath = tasksQueuePath + targetTaskKey; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
logger.info("consume task {}", taskIdPath); |
|
|
|
|
List<String> taskslist = getTasksListFromTreeSet(tasksNum, taskTreeSet); |
|
|
|
|
|
|
|
|
|
String[] vals = targetTaskKey.split(Constants.UNDERLINE); |
|
|
|
|
logger.info("consume tasks: {},there still have {} tasks need to be executed", Arrays.toString(taskslist.toArray()), size - taskslist.size()); |
|
|
|
|
|
|
|
|
|
if(remove){ |
|
|
|
|
removeNode(key, targetTaskKey); |
|
|
|
|
} |
|
|
|
|
logger.info("consume task: {},there still have {} tasks need to be executed", vals[vals.length - 1], size - 1); |
|
|
|
|
return targetTaskKey; |
|
|
|
|
return taskslist; |
|
|
|
|
}else{ |
|
|
|
|
logger.error("should not go here, task queue poll error, please check!"); |
|
|
|
|
} |
|
|
|
|
Thread.sleep(Constants.SLEEP_TIME_MILLIS); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
} catch (Exception e) { |
|
|
|
|
logger.error("add task to tasks queue exception",e); |
|
|
|
|
} |
|
|
|
|
return null; |
|
|
|
|
return new ArrayList<String>(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* get task list from tree set |
|
|
|
|
* |
|
|
|
|
* @param tasksNum |
|
|
|
|
* @param taskTreeSet |
|
|
|
|
*/ |
|
|
|
|
public List<String> getTasksListFromTreeSet(int tasksNum, Set<String> taskTreeSet) { |
|
|
|
|
Iterator<String> iterator = taskTreeSet.iterator(); |
|
|
|
|
int j = 0; |
|
|
|
|
List<String> taskslist = new ArrayList<>(tasksNum); |
|
|
|
|
while(iterator.hasNext()){ |
|
|
|
|
if(j++ < tasksNum){ |
|
|
|
|
String task = iterator.next(); |
|
|
|
|
taskslist.add(task); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
return taskslist; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@Override |
|
|
|
|
public void removeNode(String key, String nodeValue){ |
|
|
|
|
|
|
|
|
|
|
easyscheduler
5 years ago
committed by
GitHub