Shulou(Shulou.com)06/02 Report--

In this issue, the editor will bring you about how to deeply study the source code of Ali sentinel. The article is rich in content and analyzes and narrates it from a professional point of view. I hope you can get something after reading this article.

1. Ali sentinel source code research in-depth 1.1. Preface

According to my current practice, current restriction and downgrade rules don't seem to work together. I don't know why. Let's continue to explore.

1.2. Source code 1.2.1. Construction of flow control and downgrade monitoring

First of all, as far as the client is concerned, I am concerned with the code SphU.entry I wrote, which is obviously a very critical method. The following figure shows that the main flow of Sentinel work is included in the above method, through chain calls, through the establishment of tree structure, preservation of statistical cluster points, exception log records, real-time data statistics, load protection, rights authentication, flow control, circuit breaker degradation and other Slot.

The entry into the chained method is the CtSph class, within the curly braces of the try method.

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);} 1.2.2. How is the client notified when modifying console rules?

Look at the HttpEventTask class in the sentinel-transport-simple-http package, which starts a thread and is used as a socket connection. The console notifies the client through the socket request to update the client rule. The core code for changing the rule is as follows

/ / Find the matching command handler. CommandHandler commandHandler = SimpleHttpCommandCenter.getHandler (commandName); if (commandHandler! = null) {CommandResponse response = commandHandler.handle (request); handleResponse (response, printWriter, outputStream);} else {/ / No matching command handler. BadRequest (printWriter, "Unknown command `" + commandName +'`);}

Update rules when commandName is setRules in command mode

1.2.3. Since it uses socket to establish a connection, why not use netty?

With this doubt, I wanted to look for it in issues, but I suddenly found a sentinel-transport-netty-http in its source code, which is at the same level as sentinel-transport-simple-http, the official example uses simple-http, but obviously it also has netty-http, so I replaced it with netty-http, and the effect is the same as before. As for the improvement in efficiency, I don't know.

1.2.4. How are traffic rules checked?

The rule check class is FlowRuleChecker. In the core core package, the core check method is as follows

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);} 1.2.5. How to judge the downgrade of fuse?

The judgment class is DegradeRuleManager. In the core core package, the core content is as follows. If you are interested, you can see the following passCheck for yourself.

Public static void checkDegrade (ResourceWrapper resource, Context context, DefaultNode node, int count) throws BlockException {Set rules = degradeRules.get (resource.getName ()); if (rules = = null) {return;} for (DegradeRule rule: rules) {if (! rule.passCheck (context, node, count)) {throw new DegradeException (rule.getLimitApp (), rule) } 1.2.6. Where is the default chain built?

The core class is DefaultSlotChainBuilder, and the following slot is built

Public class DefaultSlotChainBuilder implements SlotChainBuilder {@ Override public ProcessorSlotChain build () {ProcessorSlotChain chain = new DefaultProcessorSlotChain (); chain.addLast (new NodeSelectorSlot ()); chain.addLast (new ClusterBuilderSlot ()); chain.addLast (new LogSlot ()); chain.addLast (new StatisticSlot ()); chain.addLast (new SystemSlot ()); chain.addLast (new AuthoritySlot ()); chain.addLast (new FlowSlot ()) Chain.addLast (new DegradeSlot ()); return chain;} 1.2.7. Now that we know how it builds chained processing nodes, how can we rebuild it ourselves?

The construction method in the discovery class SlotChainProvider is as follows

Private static void resolveSlotChainBuilder () {List list = new ArrayList (); boolean hasOther = false; for (SlotChainBuilder builder: LOADER) {if (builder.getClass ()! = DefaultSlotChainBuilder.class) {hasOther = true; list.add (builder);}} if (hasOther) {builder = list.get (0) } else {/ / No custom builder, using default. Builder = new DefaultSlotChainBuilder ();} RecordLog.info ("[SlotChainProvider] Global slot chain builder resolved:" + builder.getClass (). GetCanonicalName ());}

In other words, if we add other non-default implementations to LOADER to replace the original DefaultSlotChainBuilder, how did LOADER come from? Look at the code, the following global variable, that is, the implementation class that needs to be customized to implement the SlotChainBuilder interface

Private static final ServiceLoader LOADER = ServiceLoader.load (SlotChainBuilder.class); 1.2.8. How to implement the SlotChainBuilder interface?

One thing to note here is that it uses ServiceLoader, that is, SPI, the full name of Service Provider Interface, and it requires specific cooperation to load it, such as my custom implementation of a Slot.

/ * * @ author laoliangliang * @ date, 2019-7-25 14:13 * / public class MySlotChainBuilder implements SlotChainBuilder {@ Override public ProcessorSlotChain build () {ProcessorSlotChain chain = new DefaultProcessorSlotChain (); chain.addLast (new NodeSelectorSlot ()); chain.addLast (new ClusterBuilderSlot ()); chain.addLast (new LogSlot ()); chain.addLast (new StatisticSlot ()); chain.addLast (new SystemSlot ()) Chain.addLast (new AuthoritySlot ()); chain.addLast (new FlowSlot ()); chain.addLast (new DegradeSlot ()); / / Custom chain.addLast (new CarerSlot ()); return chain }} / * * @ author laoliangliang * @ date, 2019-7-25 14:15 * / @ Slf4jpublic class CarerSlot extends AbstractLinkedProcessorSlot {@ Override public void entry (Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, boolean prioritized, Object... Args) throws Throwable {log.info (JSON.toJSONString (resourceWrapper)); fireEntry (context, resourceWrapper, node, count, prioritized, args);} @ Override public void exit (Context context, ResourceWrapper resourceWrapper, int count, Object...) Args) {fireExit (context, resourceWrapper, count, args);}}

I have customized CarerSlot here, can it be loaded into it? In fact, it is not enough. We need to build such a file in META-INF/services/com.alibaba.csp.sentinel.slotchain.SlotChainBuilder, which is as follows.

Well, after this configuration, it can read that our custom implementation class replaces its original class.

1.2.9. Where is the initial initialization phase of the command mode?

Anyone who has used sentinel will feel that the sentinel client will be officially initialized only when the first request for sentinel monitoring comes, so where should this initialization step be?

By constantly reversing the initial initialization of the above command mode, I found the initial initialization as follows

Public class Env {public static final Sph sph = new CtSph (); static {/ / If init fails, the process will exit. InitExecutor.doInit ();}}

Do you feel familiar? DoInit is the starting point for a lot of initialization. When Env is called, static code blocks will be run, so it is only possible when sph is called.

As long as you debug the source code of my first SphU.entry above, you will find that as soon as this method enters, it will first get the sph of Env, and then call entry, so the initialization will also find the place where SphU.entry is called for the first time, or the notes you use without this, there is also this method in it.

Public static Entry entry (String name) throws BlockException {return Env.sph.entry (name, EntryType.OUT, 1, OBJECTS0);} 1.2.10. How does the annotation achieve fuse degradation?

This is actually relatively easy to understand. Since fuse degradation can be achieved through SphU.entry packaging, wrapping the code method in the form of annotations should be relatively easy, so where is it implemented and configured?

If you have read my previous article, you should see that there are the following configurations

@ Bean public SentinelResourceAspect sentinelResourceAspect () {pushlish (); return new SentinelResourceAspect ();}

Obviously, the annotation section is injected in the form of spring annotations. I think this is a more elegant way of injection. Click on it and you can see the following.

@ Pointcut ("@ annotation (com.alibaba.csp.sentinel.annotation.SentinelResource)") public void sentinelResourceAnnotationPointcut () {}

The @ SentinelResource annotation has been processed

1.2.11. What is direct failure?

This is easy to understand. If the qps exceeds the set value, it fails directly.

1.2.12. What is waiting in line?

This seems to be easy to understand literally, but once you click this option, there is a parameter below

So there is a timeout for this queue. After reaching the peak, it passes at a uniform speed. The leaky bucket algorithm and flow control diagram are adopted.

1.2.13. What is slow start mode?

The following is the core algorithm, Warm Up mode does not look at the details of the algorithm, look at its Chinese description should be able to understand what is going on; the so-called slow start mode requires that the growth rate of QPS requests of the system cannot exceed a certain rate, otherwise more than part of the requests will be suppressed and fail, in order to avoid the entry of requests with large traffic as soon as they start, which will lead to a sudden outage of the card owner or directly into the circuit breaker.

@ Override public boolean canPass (Node node, int acquireCount, boolean prioritized) {long passQps = (long) node.passQps (); long previousQps = (long) node.previousPassQps (); syncToken (previousQps); / / start calculating its slope / / if you enter the warning line, start adjusting his qps long restToken = storedTokens.get () If (restToken > = warningToken) {long aboveToken = restToken-warningToken; / / consumes faster than warning, but slower than / / current interval = restToken*slope+1/count double warningQps = Math.nextUp (1.0 / (aboveToken * slope+ 1.0 / count); if (passQps + acquireCount)

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.