1. sentinel源码工程搭建

环境准备
在nacos的官网介绍中，sentinel源码运行，需要的java运行环境有：
JDK 1.8+
Maven 3.2+
源码构建
源码下载
从github上，下载sentinel的源码到本地；
https://github.com/alibaba/Sentinel/releases

导入idea工程
1.将lg-parent工程放入sentinel源码中导入idea
2. idea工程目录
Sentinel源码分析 - 图1
3.工程启动
进入到sentinel-dashboard模块下，启动该模块下的
com.alibaba.csp.sentinel.dashboard.DashboardApplication类。
Sentinel源码分析 - 图2
sentinel项目结构
先来看下整个sentinel项目结构
sentinel-core核心模块，限流、降级、系统保护等都在这里实现
sentinel-dashboard控制台模块，可以对连接上的sentinel客户端实现可视化的管理
sentinel-transport传输模块，提供了基本的监控服务端和客户端的API接口，以及一些基于不同库的实现
sentinel-extension扩展模块，主要对DataSource进行了部分扩展实现
sentinel-adapter适配器模块，主要实现了对一些常见框架的适配
sentinel-demo样例模块，可参考怎么使用sentinel进行限流、降级等
sentinel-benchmark基准测试模块，对核心代码的精确性提供基准测试

2. sentinel源码

2.1客户端服务注册

2.1.1客户端服务注册流程分析

Sentinel源码分析 - 图3

2.1.2主要源码分析

1.导入sentinel的起步依赖后,会加载spring-cloud-alibaba-sentinel-2.1.0.RELEASE.jar下面的spring.factories文件,在文件中加载SentinelWebAutoConfiguration类,针对sentinel的自动配置类
Sentinel源码分析 - 图4
2. SentinelWebAutoConfiguration类中声明了一个FilterRegistrationBean,在这个方法中会创建sentinel核心的一个过滤器CommonFilter

  1. @Bean
  2.     @ConditionalOnProperty(
  3.         name = {"spring.cloud.sentinel.filter.enabled"},
  4.         matchIfMissing = true
  5.     )
  6.     public FilterRegistrationBean sentinelFilter() {
  7.         FilterRegistrationBean<Filter> registration = new FilterRegistrationBean();
  8.         // 获取sentinel的过滤器配置信息
  9.         com.alibaba.cloud.sentinel.SentinelProperties.Filter filterConfig = this.properties.getFilter();
  10.         if (filterConfig.getUrlPatterns() == null || filterConfig.getUrlPatterns().isEmpty()) {
  11.             List<String> defaultPatterns = new ArrayList();
  12.             //设置过滤器拦截路径为/*,拦截所有请求
  13.             defaultPatterns.add("/*");
  14.             filterConfig.setUrlPatterns(defaultPatterns);
  15.         }
  18.         registration.addUrlPatterns((String[])filterConfig.getUrlPatterns().toArray(new String[0]));
  19.         //创建CommonFilter过滤器
  20.         Filter filter = new CommonFilter();
  21.         registration.setFilter(filter);
  22.         registration.setOrder(filterConfig.getOrder());
  23.         registration.addInitParameter("HTTP_METHOD_SPECIFY", String.valueOf(this.properties.getHttpMethodSpecify()));
  24.         log.info("[Sentinel Starter] register Sentinel CommonFilter with urlPatterns: {}.", filterConfig.getUrlPatterns());
  25.         return registration;
  26.     }

  1. CommonFilter过滤器中doFilter方法会拦截所有请求 ```java @Override public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)

    1.      throws IOException, ServletException {
    2.  HttpServletRequest sRequest = (HttpServletRequest) request;
    3.  Entry urlEntry = null;
    4.  Entry httpMethodUrlEntry = null;
    6.  try {
    7.      //获取请求路径
    8.      String target = FilterUtil.filterTarget(sRequest);
    9.      // Clean and unify the URL.
    10.      // For REST APIs, you have to clean the URL (e.g. `/foo/1` and `/foo/2` -> `/foo/:id`), or
    11.      // the amount of context and resources will exceed the threshold.
    12.      UrlCleaner urlCleaner = WebCallbackManager.getUrlCleaner();
    13.      if (urlCleaner != null) {
    14.          target = urlCleaner.clean(target);
    15.      }
    17.      // If you intend to exclude some URLs, you can convert the URLs to the empty string ""
    18.      // in the UrlCleaner implementation.
    19.      if (!StringUtil.isEmpty(target)) {
    20.          // Parse the request origin using registered origin parser.
    21.          String origin = parseOrigin(sRequest);
    22.          ContextUtil.enter(WebServletConfig.WEB_SERVLET_CONTEXT_NAME, origin);
    23.          //资源检查--限流/熔断等
    24.          urlEntry = SphU.entry(target, EntryType.IN);
    25.          // Add method specification if necessary
    26.          if (httpMethodSpecify) {
    27.              httpMethodUrlEntry = SphU.entry(sRequest.getMethod().toUpperCase() + COLON + target,
    28.                      EntryType.IN);
    29.          }
    30.      }
    31.      chain.doFilter(request, response);
    32.  } catch (BlockException e) {
    33.      HttpServletResponse sResponse = (HttpServletResponse) response;
    34.      // Return the block page, or redirect to another URL.
    35.      WebCallbackManager.getUrlBlockHandler().blocked(sRequest, sResponse, e);
    36.  } catch (IOException | ServletException | RuntimeException e2) {
    37.      Tracer.traceEntry(e2, urlEntry);
    38.      Tracer.traceEntry(e2, httpMethodUrlEntry);
    39.      throw e2;
    40.  } finally {
    41.      if (httpMethodUrlEntry != null) {
    42.          httpMethodUrlEntry.exit();
    43.      }
    44.      if (urlEntry != null) {
    45.          urlEntry.exit();
    46.      }
    47.      ContextUtil.exit();
    48.  }

    }

  1. 4. SphU.entry(target, EntryType.IN)方法进行资源初始化,限流 熔断等操作
  2. ```java
  3.  public static Entry entry(String name, EntryType type) throws BlockException {
  4.      // Env类中有静态方法会被调用
  5.         return Env.sph.entry(name, type, 1, OBJECTS0);
  6.     }

  1. Env类 ```java public class Env {

    public static final Sph sph = new CtSph();

    static {

    1.  // If init fails, the process will exit.
    2.  InitExecutor.doInit();

    }

}

  1. 6. doInit方法
  2. ```java
  3.   public static void doInit() {
  4.       //判断是否是第一次初始化,不是则直接返回
  5.         if (!initialized.compareAndSet(false, true)) {
  6.             return;
  7.         }
  8.         try {
  9.             //此处去加载"META-INF/services/"目录下所配置的所有实现了InitFunc接口的类
  10.             ServiceLoader<InitFunc> loader = ServiceLoader.load(InitFunc.class);
  11.             List<OrderWrapper> initList = new ArrayList<OrderWrapper>();
  12.             for (InitFunc initFunc : loader) {
  13.                 //将加载完的所有实现类排序
  14.                 insertSorted(initList, initFunc);
  15.             }
  16.             for (OrderWrapper w : initList) {
  17.                 //执行每个InitFunc实现类的init()方法,init()方法又会去加载其它所需资源
  18.                 w.func.init();
  19.                 RecordLog.info(String.format("[InitExecutor] Executing %s with order %d",
  20.                     w.func.getClass().getCanonicalName(), w.order));
  21.             }
  22.         } catch (Exception ex) {
  24.         } catch (Error error) {
  26.         }
  27.     }

  1. w.func.init()方法会执行每个InitFunc实现类的init()方法,其中有一个实现类 HeartbeatSenderInitFunc完成客户端服务心跳发送 ```java /**

    • 心跳信号发送器的全局初始化类 *

    • @author Eric Zhao */ @InitOrder(-1) public class HeartbeatSenderInitFunc implements InitFunc {

      private ScheduledExecutorService pool = null;

      private void initSchedulerIfNeeded() { if (pool == null) {

      1.    pool = new ScheduledThreadPoolExecutor(2,
      2.        new NamedThreadFactory("sentinel-heartbeat-send-task", true),
      3.        new DiscardOldestPolicy());

      } }

      @Override public void init() { HeartbeatSender sender = HeartbeatSenderProvider.getHeartbeatSender(); if (sender == null) {

      1.    RecordLog.warn("[HeartbeatSenderInitFunc] WARN: No HeartbeatSender loaded");
      2.    return;

      }

      initSchedulerIfNeeded(); //设置心跳任务发送间隔时间 默认10s发送一次 long interval = retrieveInterval(sender); setIntervalIfNotExists(interval); //启动心跳任务 scheduleHeartbeatTask(sender, interval); }

      private boolean isValidHeartbeatInterval(Long interval) { return interval != null && interval > 0; }

      private void setIntervalIfNotExists(long interval) { SentinelConfig.setConfig(TransportConfig.HEARTBEAT_INTERVAL_MS, String.valueOf(interval)); }

      long retrieveInterval(/@NonNull/ HeartbeatSender sender) { Long intervalInConfig = TransportConfig.getHeartbeatIntervalMs(); if (isValidHeartbeatInterval(intervalInConfig)) {

      1.    RecordLog.info("[HeartbeatSenderInitFunc] Using heartbeat interval "
      2.        + "in Sentinel config property: " + intervalInConfig);
      3.    return intervalInConfig;

      } else {

      1.    long senderInterval = sender.intervalMs();
      2.    RecordLog.info("[HeartbeatSenderInit] Heartbeat interval not configured in "
      3.        + "config property or invalid, using sender default: " + senderInterval);
      4.    return senderInterval;

      } }

      private void scheduleHeartbeatTask(/@NonNull/ final HeartbeatSender sender, /@Valid/ long interval) { pool.scheduleAtFixedRate(new Runnable() {

      1.    @Override
      2.    public void run() {
      3.        try {
      4.            //发送心跳
      5.            sender.sendHeartbeat();
      6.        } catch (Throwable e) {
      7.            RecordLog.warn("[HeartbeatSender] Send heartbeat error", e);
      8.        }
      9.    }

      }, 5000, interval, TimeUnit.MILLISECONDS); RecordLog.info(“[HeartbeatSenderInit] HeartbeatSender started: “

      1.    + sender.getClass().getCanonicalName());

      } }

  1. 8. sender.sendHeartbeat();方法<br />![](https://cdn.nlark.com/yuque/0/2021/png/12497888/1618667042566-c2c139ce-4b67-4682-9dfc-4b29eb09b05d.png#clientId=uba14a92f-5cad-4&crop=0&crop=0&crop=1&crop=1&from=paste&height=538&id=ue335e04a&margin=%5Bobject%20Object%5D&originHeight=1076&originWidth=1584&originalType=binary&ratio=1&rotation=0&showTitle=false&size=228386&status=done&style=none&taskId=u73ba7ca5-febd-481d-a155-c309bd3c0cb&title=&width=792)
  3. 9.通过发送/registry/machine最终会到达sentinel服务的MachineRegistryController的receiveHeartBeat方法
  4. ```java
  5. /*
  6.  * Copyright 1999-2018 Alibaba Group Holding Ltd.
  7.  *
  8.  * Licensed under the Apache License, Version 2.0 (the "License");
  9.  * you may not use this file except in compliance with the License.
  10.  * You may obtain a copy of the License at
  11.  *
  12.  *      http://www.apache.org/licenses/LICENSE-2.0
  13.  *
  14.  * Unless required by applicable law or agreed to in writing, software
  15.  * distributed under the License is distributed on an "AS IS" BASIS,
  16.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  17.  * See the License for the specific language governing permissions and
  18.  * limitations under the License.
  19.  */
  20. package com.alibaba.csp.sentinel.dashboard.controller;
  22. import com.alibaba.csp.sentinel.dashboard.discovery.AppManagement;
  23. import com.alibaba.csp.sentinel.util.StringUtil;
  25. import com.alibaba.csp.sentinel.dashboard.discovery.MachineDiscovery;
  26. import com.alibaba.csp.sentinel.dashboard.discovery.MachineInfo;
  27. import com.alibaba.csp.sentinel.dashboard.domain.Result;
  28. import org.slf4j.Logger;
  29. import org.slf4j.LoggerFactory;
  30. import org.springframework.beans.factory.annotation.Autowired;
  31. import org.springframework.http.MediaType;
  32. import org.springframework.stereotype.Controller;
  33. import org.springframework.web.bind.annotation.RequestMapping;
  34. import org.springframework.web.bind.annotation.RequestParam;
  35. import org.springframework.web.bind.annotation.ResponseBody;
  37. /*
  38. * 注册应用处理器
  39. * */
  40. @Controller
  41. @RequestMapping(value = "/registry", produces = MediaType.APPLICATION_JSON_VALUE)
  42. public class MachineRegistryController {
  44.     private final Logger logger = LoggerFactory.getLogger(MachineRegistryController.class);
  46.     @Autowired
  47.     private AppManagement appManagement;
  48.     /**
  49.      * 注册应用信息
  50.      * */
  51.     @ResponseBody
  52.     @RequestMapping("/machine")
  53.     public Result<?> receiveHeartBeat(String app, @RequestParam(value = "app_type", required = false, defaultValue = "0") Integer appType, Long version, String v, String hostname, String ip, Integer port) {
  54.         if (app == null) {
  55.             app = MachineDiscovery.UNKNOWN_APP_NAME;
  56.         }
  57.         if (ip == null) {
  58.             return Result.ofFail(-1, "ip can't be null");
  59.         }
  60.         if (port == null) {
  61.             return Result.ofFail(-1, "port can't be null");
  62.         }
  63.         if (port == -1) {
  64.             logger.info("Receive heartbeat from " + ip + " but port not set yet");
  65.             return Result.ofFail(-1, "your port not set yet");
  66.         }
  67.         String sentinelVersion = StringUtil.isEmpty(v) ? "unknown" : v;
  68.         version = version == null ? System.currentTimeMillis() : version;
  69.         try {
  70.             MachineInfo machineInfo = new MachineInfo();
  71.             machineInfo.setApp(app);
  72.             machineInfo.setAppType(appType);
  73.             machineInfo.setHostname(hostname);
  74.             machineInfo.setIp(ip);
  75.             machineInfo.setPort(port);
  76.             machineInfo.setHeartbeatVersion(version);
  77.             machineInfo.setLastHeartbeat(System.currentTimeMillis());
  78.             machineInfo.setVersion(sentinelVersion);
  79.             // 将接受到的应用信息添加到应用程序管理appManagement
  80.             appManagement.addMachine(machineInfo);
  81.             return Result.ofSuccessMsg("success");
  82.         } catch (Exception e) {
  83.             logger.error("Receive heartbeat error", e);
  84.             return Result.ofFail(-1, e.getMessage());
  85.         }
  86.     }
  87. }

2.2 客户端请求处理

在sentinel中,一些数据存储,限流规则等都是在客户端存储的,那么客户端是怎么处理sentinel发送过来的请求呢?

2.2.1客户端请求处理流程分析

Sentinel源码分析 - 图5

2.2.1 主要源码跟踪

  1. w.func.init()方法会执行每个InitFunc实现类的init()方法,其中有一个实现类CommandCenterInitFunc完成sentinel服务端发送过来的请求相关操作
    Sentinel源码分析 - 图6
    2. commandCenter.beforeStart()注册处理器,会将所有的处理器(定义在业务方法中指定熔断或降级的处理器类)进行注册,以key-value的形式存入handlerMap中
    Sentinel源码分析 - 图7
    3. commandCenter.start();启动命令中心 ```java @Override public void start() throws Exception {

    1.  int nThreads = Runtime.getRuntime().availableProcessors();
    2.  this.bizExecutor = new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS,
    3.      new ArrayBlockingQueue<Runnable>(10),
    4.      new NamedThreadFactory("sentinel-command-center-service-executor"),
    5.      new RejectedExecutionHandler() {
    6.          @Override
    7.          public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
    8.              CommandCenterLog.info("EventTask rejected");
    9.              throw new RejectedExecutionException();
    10.          }
    11.      });
    13.  Runnable serverInitTask = new Runnable() {
    14.      int port;
    16.      {
    17.          try {
    18.              //从配置文件中获取端口,如果没有配置设置默认端口8719
    19.              port = Integer.parseInt(TransportConfig.getPort());
    20.          } catch (Exception e) {
    21.              port = DEFAULT_PORT;
    22.          }
    23.      }
    25.      @Override
    26.      public void run() {
    27.          boolean success = false;
    28.          // 获取可用的端口用以创建一个ServerSocket 如果端口被占用 则重试三次 每次端口号+1
    29.          ServerSocket serverSocket = getServerSocketFromBasePort(port);
    31.          if (serverSocket != null) {
    32.              CommandCenterLog.info("[CommandCenter] Begin listening at port " + serverSocket.getLocalPort());
    33.              socketReference = serverSocket;
    34.              // 在主线程中启动ServerThread用以接收socket请求
    35.              executor.submit(new ServerThread(serverSocket));
    36.              success = true;
    37.              port = serverSocket.getLocalPort();
    38.          } else {
    39.              CommandCenterLog.info("[CommandCenter] chooses port fail, http command center will not work");
    40.          }
    42.          if (!success) {
    43.              port = PORT_UNINITIALIZED;
    44.          }
    46.          TransportConfig.setRuntimePort(port);
    47.          executor.shutdown();
    48.      }
    50.  };
    52.  new Thread(serverInitTask).start();

    }

    /**

    • Get a server socket from an available port from a base port.
    • Increasing on port number will occur when the port has already been used. *
    • @param basePort base port to start
    • @return new socket with available port */ private static ServerSocket getServerSocketFromBasePort(int basePort) { int tryCount = 0; while (true) {
      1.  try {
      2.      //如果发现端口占用情况,则尝试3次,每次端口号加1
      3.      ServerSocket server = new ServerSocket(basePort + tryCount / 3, 100);
      4.      server.setReuseAddress(true);
      5.      return server;
      6.  } catch (IOException e) {
      7.      tryCount++;
      8.      try {
      9.          TimeUnit.MILLISECONDS.sleep(30);
      10.      } catch (InterruptedException e1) {
      11.          break;
      12.      }
      13.  }
      } return null; }
  1. ServerThread: 客户端监听到来自服务端的连接 创建一个业务线程去处理
  2. ```java
  3.     class ServerThread extends Thread {
  5.         private ServerSocket serverSocket;
  7.         ServerThread(ServerSocket s) {
  8.             this.serverSocket = s;
  9.             setName("sentinel-courier-server-accept-thread");
  10.         }
  12.         @Override
  13.         public void run() {
  14.             while (true) {
  15.                 Socket socket = null;
  16.                 try {
  17.                     //Socket监听
  18.                     socket = this.serverSocket.accept();
  19.                     setSocketSoTimeout(socket);
  20.                     // 将接收到的socket封装到HttpEventTask中由业务线程去处理 主线程不处理
  21.                     HttpEventTask eventTask = new HttpEventTask(socket);
  22.                     bizExecutor.submit(eventTask);
  23.                 } catch (Exception e) {
  24.                     CommandCenterLog.info("Server error", e);
  25.                     if (socket != null) {
  26.                         try {
  27.                             socket.close();
  28.                         } catch (Exception e1) {
  29.                             CommandCenterLog.info("Error when closing an opened socket", e1);
  30.                         }
  31.                     }
  32.                     try {
  33.                         // In case of infinite log.
  34.                         Thread.sleep(10);
  35.                     } catch (InterruptedException e1) {
  36.                         // Indicates the task should stop.
  37.                         break;
  38.                     }
  39.                 }
  40.             }
  41.         }
  42.     }

  1. HttpEventTask类处理sentinel发送过来的请求信息
    Sentinel源码分析 - 图8

  2. commandHandler.handle(request)处理请求,例如sentinel发送过来的是/setRules,则调用ModifyRulesCommandHandler

    1. @CommandMapping(name = "setRules", desc = "modify the rules, accept param: type={ruleType}&data={ruleJson}")
    2. public class ModifyRulesCommandHandler implements CommandHandler<String> {
    4.  @Override
    5.  public CommandResponse<String> handle(CommandRequest request) {
    6.      //获取规则类型
    7.      String type = request.getParam("type");
    8.      // rule data in get parameter
    9.      //获取规则数据
    10.      String data = request.getParam("data");
    11.      if (StringUtil.isNotEmpty(data)) {
    12.          try {
    13.              data = URLDecoder.decode(data, "utf-8");
    14.          } catch (Exception e) {
    15.              RecordLog.info("Decode rule data error", e);
    16.              return CommandResponse.ofFailure(e, "decode rule data error");
    17.          }
    18.      }
    20.      RecordLog.info(String.format("Receiving rule change (type: %s): %s", type, data));
    22.      String result = "success";
    24.      //限流
    25.      if (FLOW_RULE_TYPE.equalsIgnoreCase(type)) {
    26.          List<FlowRule> flowRules = JSONArray.parseArray(data, FlowRule.class);
    27.          FlowRuleManager.loadRules(flowRules);
    28.          if (!writeToDataSource(getFlowDataSource(), flowRules)) {
    29.              result = WRITE_DS_FAILURE_MSG;
    30.          }
    31.          return CommandResponse.ofSuccess(result);
    32.      } else if (AUTHORITY_RULE_TYPE.equalsIgnoreCase(type)) {//授权
    33.          List<AuthorityRule> rules = JSONArray.parseArray(data, AuthorityRule.class);
    34.          AuthorityRuleManager.loadRules(rules);
    35.          if (!writeToDataSource(getAuthorityDataSource(), rules)) {
    36.              result = WRITE_DS_FAILURE_MSG;
    37.          }
    38.          return CommandResponse.ofSuccess(result);
    39.      } else if (DEGRADE_RULE_TYPE.equalsIgnoreCase(type)) {//熔断
    40.          List<DegradeRule> rules = JSONArray.parseArray(data, DegradeRule.class);
    41.          DegradeRuleManager.loadRules(rules);
    42.          if (!writeToDataSource(getDegradeDataSource(), rules)) {
    43.              result = WRITE_DS_FAILURE_MSG;
    44.          }
    45.          return CommandResponse.ofSuccess(result);
    46.      } else if (SYSTEM_RULE_TYPE.equalsIgnoreCase(type)) {//系统规则
    47.          List<SystemRule> rules = JSONArray.parseArray(data, SystemRule.class);
    48.          SystemRuleManager.loadRules(rules);
    49.          if (!writeToDataSource(getSystemSource(), rules)) {
    50.              result = WRITE_DS_FAILURE_MSG;
    51.          }
    52.          return CommandResponse.ofSuccess(result);
    53.      }
    54.      return CommandResponse.ofFailure(new IllegalArgumentException("invalid type"));
    55.  }
    56. }

    6.假如type是限流则调用FlowRuleManager.loadRules(flowRules)去加载限流规则
    Sentinel源码分析 - 图9
    7. FlowPropertyListener类的configUpdate方法
    Sentinel源码分析 - 图10
    8. buildFlowRuleMap方法
    Sentinel源码分析 - 图11
    Sentinel源码分析 - 图12

  3. generateRater(rule)方法
    Sentinel源码分析 - 图13
    Sentinel源码分析 - 图14

2.3 sentinel限流分析

在刚才源码中我们知道客户端是如何处理sentinel发送过来的限流,那么我们看看客户端是如何完成限流操作的

2.3.1限流流程分析

Sentinel源码分析 - 图15

2.3.2主要源码跟踪
1. SphU.entry(target, EntryType.IN)代码完成限流/熔断等操作
Sentinel源码分析 - 图16
Sentinel源码分析 - 图17

2.经过调用最终会调用CtSph的entryWithPriority方法
Sentinel源码分析 - 图18
Sentinel源码分析 - 图19

  1. lookProcessChain方法构建链路
    Sentinel源码分析 - 图20
    SlotChainProvider.newSlotChain()
    Sentinel源码分析 - 图21
    builder.build() 依次在链路上加入不同的插槽
    Sentinel源码分析 - 图22
    Sentinel源码分析 - 图23
    4. chain.entry方法开启链路调用,会对链路中每个进行逐一调用,一直到到FlowSlot
    Sentinel源码分析 - 图24

  2. checkFlow限流规则检查
    Sentinel源码分析 - 图25
    ruleProvider.apply(resource.getName());
    Sentinel源码分析 - 图26

  3. canPassCheck()判断规则是否通过
    Sentinel源码分析 - 图27
    Sentinel源码分析 - 图28

  4. DefaultController默认拒接策略
    Sentinel源码分析 - 图29
    Sentinel源码分析 - 图30
    Sentinel源码分析 - 图31