序
本文主要研究一下PowerJob的DispatchStrategy
DispatchStrategy
tech/powerjob/common/enums/DispatchStrategy.java
代码语言：javascript复制@Getter
@AllArgsConstructor
public enum DispatchStrategy {
HEALTH_FIRST(1),
RANDOM(2);
private final int v;
public static DispatchStrategy of(Integer v) {
if (v == null) {
return HEALTH_FIRST;
}
for (DispatchStrategy ds : values()) {
if (v.equals(ds.v)) {
return ds;
}
}
throw new IllegalArgumentException("unknown DispatchStrategy of " v);
}
}DispatchStrategy定义了HEALTH_FIRST、RANDOM两个枚举值
getSuitableWorkers
tech/powerjob/server/remote/worker/WorkerClusterQueryService.java
代码语言：javascript复制 public List<WorkerInfo> getSuitableWorkers(JobInfoDO jobInfo) {
List<WorkerInfo> workers = Lists.newLinkedList(getWorkerInfosByAppId(jobInfo.getAppId()).values());
workers.removeIf(workerInfo -> filterWorker(workerInfo, jobInfo));
DispatchStrategy dispatchStrategy = DispatchStrategy.of(jobInfo.getDispatchStrategy());
switch (dispatchStrategy) {
case RANDOM:
Collections.shuffle(workers);
break;
case HEALTH_FIRST:
workers.sort((o1, o2) -> o2.getSystemMetrics().calculateScore() - o1.getSystemMetrics().calculateScore());
break;
default:
// do nothing
}
// 限定集群大小（0代表不限制）
if (!workers.isEmpty() && jobInfo.getMaxWorkerCount() > 0 && workers.size() > jobInfo.getMaxWorkerCount()) {
workers = workers.subList(0, jobInfo.getMaxWorkerCount());
}
return workers;
}WorkerClusterQueryService的getSuitableWorkers方法先通过getWorkerInfosByAppId获取指定appId的WorkerInfo，然后通过filterWorker进行一次过滤，最后根据dispatchStrategy来对workers进行排序，如果是RANDOM则通过Collections.shuffle(workers)随机化，如果是HEALTH_FIRST则根据systemMetrics的calculateScore结果进行排序，如果有限定maxWorkerCount则对workers进行subList，没有则返回排序后的workers
getWorkerInfosByAppId
代码语言：javascript复制 private Map<String, WorkerInfo> getWorkerInfosByAppId(Long appId) {
ClusterStatusHolder clusterStatusHolder = getAppId2ClusterStatus().get(appId);
if (clusterStatusHolder == null) {
log.warn("[WorkerManagerService] can't find any worker for app(appId={}) yet.", appId);
return Collections.emptyMap();
}
return clusterStatusHolder.getAllWorkers();
}
public Map<Long, ClusterStatusHolder> getAppId2ClusterStatus() {
return WorkerClusterManagerService.getAppId2ClusterStatus();
} getWorkerInfosByAppId通过WorkerClusterManagerService.getAppId2ClusterStatus()获取ClusterStatusHolder，在返回ClusterStatusHolder的getAllWorkers
filterWorker
代码语言：javascript复制 private boolean filterWorker(WorkerInfo workerInfo, JobInfoDO jobInfo) {
for (WorkerFilter filter : workerFilters) {
if (filter.filter(workerInfo, jobInfo)) {
return true;
}
}
return false;
}filterWorker方法则是遍历workerFilters直接filter
calculateScore
tech/powerjob/common/model/SystemMetrics.java
代码语言：javascript复制 public int calculateScore() {
if (score > 0) {
return score;
}
// Memory is vital to TaskTracker, so we set the multiplier factor as 2.
double memScore = (jvmMaxMemory - jvmUsedMemory) * 2;
// Calculate the remaining load of CPU. Multiplier is set as 1.
double cpuScore = cpuProcessors - cpuLoad;
// Windows can not fetch CPU load, set cpuScore as 1.
if (cpuScore > cpuProcessors) {
cpuScore = 1;
}
score = (int) (memScore cpuScore);
return score;
}SystemMetrics的calculateScore方法则是基于memScore与cpuScore来计算
WorkerFilter
代码语言：javascript复制public interface WorkerFilter {
/**
*
* @param workerInfo worker info, maybe you need to use your customized info in SystemMetrics#extra
* @param jobInfoDO job info
* @return true will remove the worker in process list
*/
boolean filter(WorkerInfo workerInfo, JobInfoDO jobInfoDO);
}WorkerFilter定义了filter接口用于过滤worker，它有3个实现类，分别是DesignatedWorkerFilter、DisconnectedWorkerFilter、SystemMetricsWorkerFilter
DesignatedWorkerFilter
tech/powerjob/server/extension/defaultimpl/workerfilter/DesignatedWorkerFilter.java
代码语言：javascript复制@Slf4j
@Component
public class DesignatedWorkerFilter implements WorkerFilter {
@Override
public boolean filter(WorkerInfo workerInfo, JobInfoDO jobInfo) {
String designatedWorkers = jobInfo.getDesignatedWorkers();
// no worker is specified, no filter of any
if (StringUtils.isEmpty(designatedWorkers)) {
return false;
}
Set<String> designatedWorkersSet = Sets.newHashSet(SJ.COMMA_SPLITTER.splitToList(designatedWorkers));
for (String tagOrAddress : designatedWorkersSet) {
if (tagOrAddress.equals(workerInfo.getTag()) || tagOrAddress.equals(workerInfo.getAddress())) {
return false;
}
}
return true;
}
}DesignatedWorkerFilter的filter方法遍历jobInfo的designatedWorkers信息，判断workerInfo的tag或者address是否匹配
DisconnectedWorkerFilter
tech/powerjob/server/extension/defaultimpl/workerfilter/DisconnectedWorkerFilter.java
代码语言：javascript复制@Slf4j
@Component
public class DisconnectedWorkerFilter implements WorkerFilter {
@Override
public boolean filter(WorkerInfo workerInfo, JobInfoDO jobInfo) {
boolean timeout = workerInfo.timeout();
if (timeout) {
log.info("[Job-{}] filter worker[{}] due to timeout(lastActiveTime={})", jobInfo.getId(), workerInfo.getAddress(), workerInfo.getLastActiveTime());
}
return timeout;
}
}DisconnectedWorkerFilter的filter方法则通过WorkerInfo的timeout方法来判断，它主要是判断当前时间与lastActiveTime的时间差是否大于WORKER_TIMEOUT_MS(
60s)
SystemMetricsWorkerFilter
tech/powerjob/server/extension/defaultimpl/workerfilter/SystemMetricsWorkerFilter.java
代码语言：javascript复制@Slf4j
@Component
public class SystemMetricsWorkerFilter implements WorkerFilter {
@Override
public boolean filter(WorkerInfo workerInfo, JobInfoDO jobInfo) {
SystemMetrics metrics = workerInfo.getSystemMetrics();
boolean filter = !metrics.available(jobInfo.getMinCpuCores(), jobInfo.getMinMemorySpace(), jobInfo.getMinDiskSpace());
if (filter) {
log.info("[Job-{}] filter worker[{}] because the {} do not meet the requirements", jobInfo.getId(), workerInfo.getAddress(), workerInfo.getSystemMetrics());
}
return filter;
}
}SystemMetricsWorkerFilter的filter方法则根据workerInfo的SystemMetrics判断可用cpu核数、内存、磁盘空间是否大于阈值
小结
DispatchStrategy定义了HEALTH_FIRST、RANDOM两个枚举值；WorkerClusterQueryService的getSuitableWorkers方法先通过getWorkerInfosByAppId获取指定appId的WorkerInfo，然后通过filterWorker进行一次过滤，最后根据dispatchStrategy来对workers进行排序，如果是RANDOM则通过Collections.shuffle(workers)随机化，如果是HEALTH_FIRST则根据systemMetrics的calculateScore结果进行排序，如果有限定maxWorkerCount则对workers进行subList，没有则返回排序后的workers。
