聊聊sentinel的SystemSlot-白红宇

聊聊sentinel的SystemSlot

阅读量：6316 次

发布时间：2019-06-22

本文共 5079 字，大约阅读时间需要 16 分钟。

序

本文主要研究一下sentinel的SystemSlot

SystemSlot

sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/system/SystemSlot.java

public class SystemSlot extends AbstractLinkedProcessorSlot
    
      {    @Override    public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, Object... args)        throws Throwable {        SystemRuleManager.checkSystem(resourceWrapper);        fireEntry(context, resourceWrapper, node, count, args);    }    @Override    public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {        fireExit(context, resourceWrapper, count, args);    }}

这里是通过SystemRuleManager.checkSystem(resourceWrapper)进行系统限流判断

SystemRuleManager

public static void checkSystem(ResourceWrapper resourceWrapper) throws BlockException {        // 确定开关开了        if (checkSystemStatus.get() == false) {            return;        }        // for inbound traffic only        if (resourceWrapper.getType() != EntryType.IN) {            return;        }        // total qps        double currentQps = Constants.ENTRY_NODE == null ? 0.0 : Constants.ENTRY_NODE.successQps();        if (currentQps > qps) {            throw new SystemBlockException(resourceWrapper.getName(), "qps");        }        // total thread        int currentThread = Constants.ENTRY_NODE == null ? 0 : Constants.ENTRY_NODE.curThreadNum();        if (currentThread > maxThread) {            throw new SystemBlockException(resourceWrapper.getName(), "thread");        }        double rt = Constants.ENTRY_NODE == null ? 0 : Constants.ENTRY_NODE.avgRt();        if (rt > maxRt) {            throw new SystemBlockException(resourceWrapper.getName(), "rt");        }        // 完全按照RT,BBR算法来        if (highestSystemLoadIsSet && getCurrentSystemAvgLoad() > highestSystemLoad) {            if (currentThread > 1 &&                currentThread > Constants.ENTRY_NODE.maxSuccessQps() * Constants.ENTRY_NODE.minRt() / 1000) {                throw new SystemBlockException(resourceWrapper.getName(), "load");            }        }    }    public static double getCurrentSystemAvgLoad() {        return statusListener.getSystemAverageLoad();    }

先判断qps，在判断总线程数、之后判断rt，最后判断系统负载有没有超过限制

StatisticNode

sentinel-core/src/main/java/com/alibaba/csp/sentinel/node/StatisticNode.java

@Override    public long successQps() {        return rollingCounterInSecond.success() / IntervalProperty.INTERVAL;    }    @Override    public int curThreadNum() {        return curThreadNum.get();    }    @Override    public long avgRt() {        long successCount = rollingCounterInSecond.success();        if (successCount == 0) {            return 0;        }        return rollingCounterInSecond.rt() / successCount;    }    @Override    public long maxSuccessQps() {        return rollingCounterInSecond.maxSuccess() * SampleCountProperty.sampleCount;    }    @Override    public long minRt() {        return rollingCounterInSecond.minRt();    }

successQps、curThreadNum、avgRt、maxSuccessQps、minRt指标在StatisticNode上进行维护

SystemStatusListener

sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/system/SystemStatusListener.java

public class SystemStatusListener implements Runnable {    volatile double currentLoad = -1;    volatile String reason = StringUtil.EMPTY;    static final int processor = ManagementFactory.getOperatingSystemMXBean().getAvailableProcessors();    public double getSystemAverageLoad() {        return currentLoad;    }    @Override    public void run() {        try {            if (!SystemRuleManager.getCheckSystemStatus()) {                return;            }            // system average load            OperatingSystemMXBean operatingSystemMXBean = ManagementFactory.getOperatingSystemMXBean();            currentLoad = operatingSystemMXBean.getSystemLoadAverage();            StringBuilder sb = new StringBuilder();            if (currentLoad > SystemRuleManager.getHighestSystemLoad()) {                sb.append("load:").append(currentLoad).append(";");                sb.append("qps:").append(Constants.ENTRY_NODE.passQps()).append(";");                sb.append("rt:").append(Constants.ENTRY_NODE.avgRt()).append(";");                sb.append("thread:").append(Constants.ENTRY_NODE.curThreadNum()).append(";");                sb.append("success:").append(Constants.ENTRY_NODE.successQps()).append(";");                sb.append("minRt:").append(Constants.ENTRY_NODE.minRt()).append(";");                sb.append("maxSuccess:").append(Constants.ENTRY_NODE.maxSuccessQps()).append(";");                RecordLog.info(sb.toString());            }        } catch (Throwable e) {            RecordLog.info("could not get system error ", e);        }    }}

系统负载是通过SystemRuleManager定时调度SystemStatusListener，通过OperatingSystemMXBean去获取

static {        checkSystemStatus.set(false);        statusListener = new SystemStatusListener();        scheduler.scheduleAtFixedRate(statusListener, 5, 1, TimeUnit.SECONDS);        currentProperty.addListener(listener);    }