最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

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
        }

        // 限定集群大?。?代表不限制)
        if (!workers.isEmpty() && jobInfo.getMaxWorkerCount() > 0 && workers.size() > jobInfo.getMaxWorkerCount()) {
            workers = workers.subList(0, jobInfo.getMaxWorkerCount());
        }
        return workers;
    }
WorkerClusterQueryService的getSuitableWorkers方法先通過(guò)getWorkerInfosByAppId獲取指定appId的WorkerInfo,然后通過(guò)filterWorker進(jìn)行一次過(guò)濾,最后根據(jù)dispatchStrategy來(lái)對(duì)workers進(jìn)行排序,如果是RANDOM則通過(guò)Collections.shuffle(workers)隨機(jī)化,如果是HEALTH_FIRST則根據(jù)systemMetrics的calculateScore結(jié)果進(jìn)行排序,如果有限定maxWorkerCount則對(duì)workers進(jìn)行subList,沒(méi)有則返回排序后的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通過(guò)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來(lái)計(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接口用于過(guò)濾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方法則通過(guò)WorkerInfo的timeout方法來(lái)判斷,它主要是判斷當(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方法先通過(guò)getWorkerInfosByAppId獲取指定appId的WorkerInfo,然后通過(guò)filterWorker進(jìn)行一次過(guò)濾,最后根據(jù)dispatchStrategy來(lái)對(duì)workers進(jìn)行排序,如果是RANDOM則通過(guò)Collections.shuffle(workers)隨機(jī)化,如果是HEALTH_FIRST則根據(jù)systemMetrics的calculateScore結(jié)果進(jìn)行排序,如果有限定maxWorkerCount則對(duì)workers進(jìn)行subList,沒(méi)有則返回排序后的workers。

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

相關(guān)文章

最新評(píng)論

仪征市| 济南市| 赤城县| 乐平市| 天峻县| 高邑县| 潞西市| 祥云县| 丹巴县| 德州市| 宁夏| 庆云县| 青岛市| 石台县| 文水县| 蓬安县| 灯塔市| 西昌市| 贺兰县| 五河县| 平罗县| 阜平县| 玉龙| 五原县| 安平县| 浦北县| 贺州市| 务川| 汾阳市| 肇州县| 霞浦县| 安达市| 梁平县| 平利县| 宣恩县| 闻喜县| 板桥市| 盐津县| 随州市| 益阳市| 武乡县|