欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

PowerJob的DispatchStrategy方法工作流程源碼解讀

 更新時(shí)間:2024年01月12日 09:34:39   作者:codecraft  
這篇文章主要為大家介紹了PowerJob的DispatchStrategy方法工作流程源碼解讀,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪

本文主要研究一下PowerJob的DispatchStrategy

DispatchStrategy

tech/powerjob/common/enums/DispatchStrategy.java

@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兩個(gè)枚舉值

getSuitableWorkers

tech/powerjob/server/remote/worker/WorkerClusterQueryService.java

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進(jìn)行一次過濾,最后根據(jù)dispatchStrategy來對(duì)workers進(jìn)行排序,如果是RANDOM則通過Collections.shuffle(workers)隨機(jī)化,如果是HEALTH_FIRST則根據(jù)systemMetrics的calculateScore結(jié)果進(jìn)行排序,如果有限定maxWorkerCount則對(duì)workers進(jìn)行subList,沒有則返回排序后的workers

getWorkerInfosByAppId

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

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

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來計(jì)算

WorkerFilter

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個(gè)實(shí)現(xiàn)類,分別是DesignatedWorkerFilter、DisconnectedWorkerFilter、SystemMetricsWorkerFilter

DesignatedWorkerFilter

tech/powerjob/server/extension/defaultimpl/workerfilter/DesignatedWorkerFilter.java

@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

@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方法來判斷,它主要是判斷當(dāng)前時(shí)間與lastActiveTime的時(shí)間差是否大于WORKER_TIMEOUT_MS(60s)

SystemMetricsWorkerFilter

tech/powerjob/server/extension/defaultimpl/workerfilter/SystemMetricsWorkerFilter.java

@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方法則根據(jù)workerInfo的SystemMetrics判斷可用cpu核數(shù)、內(nèi)存、磁盤空間是否大于閾值

小結(jié)

DispatchStrategy定義了HEALTH_FIRST、RANDOM兩個(gè)枚舉值;WorkerClusterQueryService的getSuitableWorkers方法先通過getWorkerInfosByAppId獲取指定appId的WorkerInfo,然后通過filterWorker進(jìn)行一次過濾,最后根據(jù)dispatchStrategy來對(duì)workers進(jìn)行排序,如果是RANDOM則通過Collections.shuffle(workers)隨機(jī)化,如果是HEALTH_FIRST則根據(jù)systemMetrics的calculateScore結(jié)果進(jìn)行排序,如果有限定maxWorkerCount則對(duì)workers進(jìn)行subList,沒有則返回排序后的workers。

以上就是PowerJob的DispatchStrategy方法工作流程源碼解讀的詳細(xì)內(nèi)容,更多關(guān)于PowerJob DispatchStrategy的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!

相關(guān)文章

  • Java 將PPT幻燈片轉(zhuǎn)為HTML文件的實(shí)現(xiàn)思路

    Java 將PPT幻燈片轉(zhuǎn)為HTML文件的實(shí)現(xiàn)思路

    本文以Java程序代碼為例展示如何通過格式轉(zhuǎn)換的方式將PPT幻燈片文檔轉(zhuǎn)為HTML文件,本文通過實(shí)例代碼圖文相結(jié)合給大家分享實(shí)現(xiàn)思路,需要的朋友參考下吧
    2021-06-06
  • 最新評(píng)論