Fork me on GitHub

framework/SpringCloud/sentinel/Sentinel启动流程

字数统计: 2.4k字   |   阅读时长≈ 12分

Sentinel启动流程

1 环境初始化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

/**
 * Sentinel Env. This class will trigger all initialization for Sentinel.
 *
 * <p>
 * NOTE: to prevent deadlocks, other classes' static code block or static field should
 * NEVER refer to this class.
 * </p>
 *
 * @author jialiang.linjl
 */
public class Env {
	// 这个先不看
    public static final Sph sph = new CtSph();

    static {
        // If init fails, the process will exit.
        InitExecutor.doInit();
    }

}

InitExecutor.doInit();

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

/**
    * If one {@link InitFunc} throws an exception, the init process
    * will immediately be interrupted and the application will exit.
    *
    * The initialization will be executed only once.
    */
   public static void doInit() {
       if (!initialized.compareAndSet(false, true)) {
           return;
       }
       //只有一个线程可以进行初始化
       
       try {
           // 这是通过 ServiceLoader去加载类，炫技，默认回去meta-inf/service找指定类加载
           ServiceLoader<InitFunc> loader = ServiceLoaderUtil.getServiceLoader(InitFunc.class);
           
           List<OrderWrapper> initList = new ArrayList<OrderWrapper>();
           for (InitFunc initFunc : loader) {
               RecordLog.info("[InitExecutor] Found init func: " + initFunc.getClass().getCanonicalName());
               insertSorted(initList, initFunc);
           }
           for (OrderWrapper w : initList) {
               w.func.init();
               RecordLog.info(String.format("[InitExecutor] Executing %s with order %d",
                   w.func.getClass().getCanonicalName(), w.order));
           }
       } catch (Exception ex) {
           RecordLog.warn("[InitExecutor] WARN: Initialization failed", ex);
           ex.printStackTrace();
       } catch (Error error) {
           RecordLog.warn("[InitExecutor] ERROR: Initialization failed with fatal error", error);
           error.printStackTrace();
       }
   }

spi机制

MetricCallbackInit

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

/**
 * Register callbacks for metric extension.
 *
 * @author Carpenter Lee
 * @since 1.6.1
 */
public class MetricCallbackInit implements InitFunc {
    @Override
    public void init() throws Exception {
        StatisticSlotCallbackRegistry.addEntryCallback(MetricEntryCallback.class.getCanonicalName(),
            new MetricEntryCallback());
        StatisticSlotCallbackRegistry.addExitCallback(MetricExitCallback.class.getCanonicalName(),
            new MetricExitCallback());
    }
}

Sentinal执行流程

entry

SphU

1
2
3
4

public static Entry entry(String name) throws BlockException {
        return Env.sph.entry(name, EntryType.OUT, 1, OBJECTS0);
    }

CtSph

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

    @Override
    public Entry entry(String name, EntryType type, int count, Object... args) throws BlockException {
        // 资源包装
        StringResourceWrapper resource = new StringResourceWrapper(name, type);
        return entry(resource, count, args);
    }


    public Entry entry(ResourceWrapper resourceWrapper, int count, Object... args) throws BlockException {
        return entryWithPriority(resourceWrapper, count, false, args);
    }


private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args)
        throws BlockException {
        Context context = ContextUtil.getContext();
        if (context instanceof NullContext) {
            // The {@link NullContext} indicates that the amount of context has exceeded the threshold,
            // so here init the entry only. No rule checking will be done.
            return new CtEntry(resourceWrapper, null, context);
        }

        if (context == null) {
            // Using default context.
            // 设置当前上下文
            context = InternalContextUtil.internalEnter(Constants.CONTEXT_DEFAULT_NAME);
        }

        // Global switch is close, no rule checking will do.
        if (!Constants.ON) {
            return new CtEntry(resourceWrapper, null, context);
        }
		// 获取插槽链
        ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper);

        /*
         * Means amount of resources (slot chain) exceeds {@link Constants.MAX_SLOT_CHAIN_SIZE},
         * so no rule checking will be done.
         */
        if (chain == null) {
            return new CtEntry(resourceWrapper, null, context);
        }
		// 初始化Entry
        Entry e = new CtEntry(resourceWrapper, chain, context);
        try {
            // 链表执行
            chain.entry(context, resourceWrapper, null, count, prioritized, args);
        } catch (BlockException e1) {
            //错误，资源退出
            e.exit(count, args);
            throw e1;
        } catch (Throwable e1) {
            // This should not happen, unless there are errors existing in Sentinel internal.
            RecordLog.info("Sentinel unexpected exception", e1);
        }
        return e;
    }

lookProcessChain(resourceWrapper);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38




public class DefaultSlotChainBuilder implements SlotChainBuilder {

    @Override
    public ProcessorSlotChain build() {
        ProcessorSlotChain chain = new DefaultProcessorSlotChain();

        // Note: the instances of ProcessorSlot should be different, since they are not stateless.
        List<ProcessorSlot> sortedSlotList = SpiLoader.loadPrototypeInstanceListSorted(ProcessorSlot.class);
        for (ProcessorSlot slot : sortedSlotList) {
            if (!(slot instanceof AbstractLinkedProcessorSlot)) {
                RecordLog.warn("The ProcessorSlot(" + slot.getClass().getCanonicalName() + ") is not an instance of AbstractLinkedProcessorSlot, can't be added into ProcessorSlotChain");
                continue;
            }

            chain.addLast((AbstractLinkedProcessorSlot<?>) slot);
        }

        return chain;
    }
}

//由spi

# Sentinel default ProcessorSlots
com.alibaba.csp.sentinel.slots.nodeselector.NodeSelectorSlot
com.alibaba.csp.sentinel.slots.clusterbuilder.ClusterBuilderSlot
com.alibaba.csp.sentinel.slots.logger.LogSlot
com.alibaba.csp.sentinel.slots.statistic.StatisticSlot
com.alibaba.csp.sentinel.slots.block.authority.AuthoritySlot
com.alibaba.csp.sentinel.slots.system.SystemSlot
com.alibaba.csp.sentinel.slots.block.flow.FlowSlot
com.alibaba.csp.sentinel.slots.block.degrade.DegradeSlot
    
    可以获得很多插槽
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@Override
  public void entry(Context context, ResourceWrapper resourceWrapper, Object t, int count, boolean prioritized, Object... args)
      throws Throwable {
      first.transformEntry(context, resourceWrapper, t, count, prioritized, args);
  }


 void transformEntry(Context context, ResourceWrapper resourceWrapper, Object o, int count, boolean prioritized, Object... args)
      throws Throwable {
      T t = (T)o;
      entry(context, resourceWrapper, t, count, prioritized, args);
  }


@Override
      public void entry(Context context, ResourceWrapper resourceWrapper, Object t, int count, boolean prioritized, Object... args)
          throws Throwable {
          super.fireEntry(context, resourceWrapper, t, count, prioritized, args);
      }


 @Override
  public void fireEntry(Context context, ResourceWrapper resourceWrapper, Object obj, int count, boolean prioritized, Object... args)
      throws Throwable {
      if (next != null) {
          next.transformEntry(context, resourceWrapper, obj, count, prioritized, args);
      }
  }

每个处理插槽的作用

FlowSlot

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
/*
结合从以前的slot（NodeSelectorSlot、ClusterNodeBuilderSlot和StatisticSlot）收集的运行时统计信息，FlowSlot将使用预先设置的规则来决定是否应阻止传入的请求。

{@code SphU.entry(resourceName)} 将在触发任何规则时引发{@Code FlowException}。用户可以通过捕捉{@code flowexception}自定义自己的逻辑。

一个资源可以有多个流规则。FlowSlot遍历这些规则，直到触发其中一个规则或遍历了所有规则。


每个{@link flowrule}主要由以下因素组成：等级、策略、路径。我们可以结合这些因素来达到不同的效果。

等级由{@link FlowRule}中的{@code grade}字段定义。这里，0表示线程隔离，1表示请求计数整形（QPS）。线程计数和请求计数都是在实际运行时收集的，我们可以通过以下命令查看这些统计信息：

* <pre>
 * curl http://localhost:8719/tree
 *
 * idx id    thread pass  blocked   success total aRt   1m-pass   1m-block   1m-all   exception
 * 2   abc647 0      460    46          46   1    27      630       276        897      0
 * </pre>
 
 
 * <ul>
 * <li>{@code thread} 当前正在处理资源的线程数</li>
 * <li>{@code pass} 一秒钟内传入请求的计数</li>
 * <li>{@code blocked} 一秒钟内阻止的请求数</li>
 * <li>{@code success} 对于Sentinel在一秒钟内成功处理的请求的计数</li>
 * <li>{@code RT} 请求在一秒钟内的平均响应时间</li>
 * <li>{@code total} 一秒钟内传入请求和阻止请求的总和</li>
 * <li>{@code 1m-pass} 表示一分钟内传入请求的计数</li>
 * <li>{@code 1m-block} 用于一分钟内阻止的请求的计数</li>
 * <li>{@code 1m-all} 是一分钟内传入和阻止的请求总数</li>
 * <li>{@code exception} 用于一秒钟内的业务（自定义）异常计数</li>
 * </ul>



这个阶段通常用来保护资源不被占用。如果资源需要很长时间才能完成，线程将开始占用。响应时间越长，占用的线程就越多。

除了计数器之外，还可以使用线程池或信号量来实现这一点。
Thread pool: 分配一个线程池来处理这些资源。当池中不再有空闲线程时，请求将被拒绝，而不会影响其他资源。

Semaphore: 使用信号量控制此资源中线程的并发计数。
使用线程池的好处是，它可以在超时时优雅地退出。但是它也给我们带来了上下文切换和额外线程的开销。如果传入的请求已经在一个单独的线程中提供服务，例如，一个Servlet HTTP请求，那么如果使用线程池，那么它的线程数几乎会增加一倍。


流量控制：
当QPS超过阈值时，Sentinel将采取措施控制传入的请求，并由流规则中的{@code controlBehavior}字段配置。
立即拒绝（{@codeRuleConstant.CONTROL_行为\u DEFAULT})

这是默认行为。将立即拒绝超出的请求并引发FlowException


* <li>Warmup ({@code RuleConstant.CONTROL_BEHAVIOR_WARM_UP})</li>
 * <p>
 * 如果系统负载低了一段时间，并且出现了大量的请求，那么系统可能无法同时处理所有这些请求。但是，如果我们稳定地增加传入的请求，系统就可以预热并最终能够处理所有请求。可以通过在流规则中设置字段{@code warmUpPeriodSec}来配置此预热期。
 * </p>
 
  * <li>Uniform Rate Limiting ({@code RuleConstant.CONTROL_BEHAVIOR_RATE_LIMITER})</li>
 * <p>
 *此策略严格控制请求之间的间隔。换句话说，它允许请求以稳定、统一的速率通过。

 * 该策略是漏桶的一种实现。它用于以稳定的速率处理请求，通常用于突发通信（如消息处理）。当超过系统容量的大量请求同时到达时，使用该策略的系统将以固定速率处理请求，直到所有请求都被处理或超时。
 * </p>
 * </ol>
 
*/

@SpiOrder(-2000)
public class FlowSlot extends AbstractLinkedProcessorSlot<DefaultNode> {

    private final FlowRuleChecker checker;

    public FlowSlot() {
        this(new FlowRuleChecker());
    }

    /**
     * Package-private for test.
     *
     * @param checker flow rule checker
     * @since 1.6.1
     */
    FlowSlot(FlowRuleChecker checker) {
        AssertUtil.notNull(checker, "flow checker should not be null");
        this.checker = checker;
    }

    @Override
    public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                      boolean prioritized, Object... args) throws Throwable {
        checkFlow(resourceWrapper, context, node, count, prioritized);

        fireEntry(context, resourceWrapper, node, count, prioritized, args);
    }

    void checkFlow(ResourceWrapper resource, Context context, DefaultNode node, int count, boolean prioritized)
        throws BlockException {
        checker.checkFlow(ruleProvider, resource, context, node, count, prioritized);
    }

    @Override
    public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
        fireExit(context, resourceWrapper, count, args);
    }

    
    //一个规则的提供者
    private final Function<String, Collection<FlowRule>> ruleProvider = new Function<String, Collection<FlowRule>>() {
        @Override
        public Collection<FlowRule> apply(String resource) {
            // Flow rule map should not be null.
            Map<String, List<FlowRule>> flowRules = FlowRuleManager.getFlowRuleMap();
            return flowRules.get(resource);
        }
    };
}

FlowRuleChecker

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

public void checkFlow(Function<String, Collection<FlowRule>> ruleProvider, ResourceWrapper resource,
                          Context context, DefaultNode node, int count, boolean prioritized) throws BlockException {
        if (ruleProvider == null || resource == null) {
            return;
        }
        Collection<FlowRule> rules = ruleProvider.apply(resource.getName());
        if (rules != null) {
            for (FlowRule rule : rules) {
                // 进行规则检查，不通过则抛出异常
                if (!canPassCheck(rule, context, node, count, prioritized)) {
                    throw new FlowException(rule.getLimitApp(), rule);
                }
            }
        }
    }


 public boolean canPassCheck(/*@NonNull*/ FlowRule rule, Context context, DefaultNode node, int acquireCount,
                                                    boolean prioritized) {
        String limitApp = rule.getLimitApp();
        if (limitApp == null) {
            return true;
        }

        if (rule.isClusterMode()) {
            return passClusterCheck(rule, context, node, acquireCount, prioritized);
        }

        return passLocalCheck(rule, context, node, acquireCount, prioritized);
    }

passClusterCheck

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

private static boolean passClusterCheck(FlowRule rule, Context context, DefaultNode node, int acquireCount,
                                            boolean prioritized) {
        try {
            TokenService clusterService = pickClusterService();
            if (clusterService == null) {
                return fallbackToLocalOrPass(rule, context, node, acquireCount, prioritized);
            }
            long flowId = rule.getClusterConfig().getFlowId();
            TokenResult result = clusterService.requestToken(flowId, acquireCount, prioritized);
            return applyTokenResult(result, rule, context, node, acquireCount, prioritized);
            // If client is absent, then fallback to local mode.
        } catch (Throwable ex) {
            RecordLog.warn("[FlowRuleChecker] Request cluster token unexpected failed", ex);
        }
        // Fallback to local flow control when token client or server for this rule is not available.
        // If fallback is not enabled, then directly pass.
        return fallbackToLocalOrPass(rule, context, node, acquireCount, prioritized);
    }

passLocalCheck

1
2
3
4
5
6
7
8
9
private static boolean passLocalCheck(FlowRule rule, Context context, DefaultNode node, int acquireCount,
                                          boolean prioritized) {
        Node selectedNode = selectNodeByRequesterAndStrategy(rule, context, node);
        if (selectedNode == null) {
            return true;
        }

        return rule.getRater().canPass(selectedNode, acquireCount, prioritized);
    }
赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

主要涉及技术：
Java后端开发
开源爱好者、Linux

联系QQ:2960571342

很惭愧

只做了一点微小的工作
谢谢大家