关于SegmentFault:Soul源码阅读02devide插件负载均衡权重分析

30次阅读

共计 6917 个字符，预计需要花费 18 分钟才能阅读完成。

指标

  • 运行 examples 上面的 http 服务
  • 学习文档，联合 divde 插件，发动 http 申请 soul 网关，体验 http 代理

http 服务的相干依赖及配置

  • 引入 http 的代理插件

在 soul-bootstrap 工程下的 pom.xml下引入如下依赖

 <!--if you use http proxy start this-->
 <dependency>
     <groupId>org.dromara</groupId>
     <artifactId>soul-spring-boot-starter-plugin-divide</artifactId>
     <version>${project.version}</version>
 </dependency>
 <dependency>
     <groupId>org.dromara</groupId>
     <artifactId>soul-spring-boot-starter-plugin-httpclient</artifactId>
     <version>${project.version}</version>
 </dependency>
  • 引入 soul 客户端（针对 SpringBoot 用户）：

在 soul-examples-http(你本人的实在服务) pom.xml 新增如下依赖:

    <dependency>
         <groupId>org.dromara</groupId>
         <artifactId>soul-spring-boot-starter-client-springmvc</artifactId>
         <version>${last.version}</version>
     </dependency>
  • 增加客户端接入配置：

在 yml 中新增如下配置：

soul:
  http:
    adminUrl: http://localhost:9095
    port: 8188
    contextPath: /myhttp
    appName: http
    full: false
    
   # adminUrl: 为你启动的 soul-admin 我的项目的 ip + 端口，留神要加 http://
   # port: 你本我的项目的启动端口
   # contextPath: 为你的这个 mvc 我的项目在 soul 网关的路由前缀，比方 /order，/product 等等，网关会依据你的这个前缀来进行路由
   # appName：你的利用名称，不配置的话，会默认取 `spring.application.name` 的值
   # full: 设置 true 代表代理你的整个服务，false 示意代理你其中某几个 controller
  • Controller中增加 @SoulSpringMvcClient 注解

将注解 Controller 类下面，外面的 path 属性则为前缀，如果含有 /** 代表你的整个接口须要被网关代理。

@RestController
@RequestMapping("/test")
@SoulSpringMvcClient(path = "/test/**")
public class HttpTestController {//controller 中所有办法被网关代理}

@RestController
@RequestMapping("/order")
@SoulSpringMvcClient(path = "/order")
public class OrderController {
    /**
      * order/save 会被网关代理, 而 /order/findById 则不会
      */
    @PostMapping("/save")
    @SoulSpringMvcClient(path = "/save" , desc = "Save order")
    public OrderDTO save(@RequestBody final OrderDTO orderDTO) {orderDTO.setName("hello world save order");
        return orderDTO;
    }

    @GetMapping("/findById")
    public OrderDTO findById(@RequestParam("id") final String id) {OrderDTO orderDTO = new OrderDTO();
        orderDTO.setId(id);
        orderDTO.setName("hello world findById");
        return orderDTO;
    }
}

devide 负载平衡权重剖析

批改 diea 启动配置，勾选Allow parallel run，容许并行启动

批改 application.yml 中端口配置，将端口改为 8189

server:
  port: 8189 # 批改端口
  address: 0.0.0.0

soul:
  http:
    adminUrl: http://localhost:9095
    port: 8189 # 批改端口
    contextPath: /http
    appName: http
    full: false

胜利启动，soul 后盾中会新注册一个 8189 服务

批改 weight 权重配置，将 8188 权重改为 100

采纳 假设性准则 ，依据工程名及文件名称，在DividePlugin.java 的doExecute办法中增加断点。应用 Postman 对网关发动一个申请，遇到断点，查看调用栈信息

始终往上找，发现申请会先进入 SoulWebHandler 类的 handle 办法里：

    @Override
    public Mono<Void> handle(@NonNull final ServerWebExchange exchange) {MetricsTrackerFacade.getInstance().counterInc(MetricsLabelEnum.REQUEST_TOTAL.getName());
        Optional<HistogramMetricsTrackerDelegate> startTimer = MetricsTrackerFacade.getInstance().histogramStartTimer(MetricsLabelEnum.REQUEST_LATENCY.getName());
        return new DefaultSoulPluginChain(plugins).execute(exchange).subscribeOn(scheduler)
                .doOnSuccess(t -> startTimer.ifPresent(time -> MetricsTrackerFacade.getInstance().histogramObserveDuration(time)));
    }

而后调用 DefaultSoulPluginChain 的 execute 办法，从 plugins 中获取 DividePlugin 插件：

    public Mono<Void> execute(final ServerWebExchange exchange) {return Mono.defer(() -> {if (this.index < plugins.size()) {
                    // 获取插件
                    SoulPlugin plugin = plugins.get(this.index++);
                    Boolean skip = plugin.skip(exchange);
                    if (skip) {
                        // 跳过插件
                        return this.execute(exchange);
                    }
                    // 调用插件的具体执行办法
                    return plugin.execute(exchange, this);
                }
                return Mono.empty();});
        }

调用了插件的具体执行办法后，进入 AbstractSoulPlugin 类的 execute 办法，这里应用了 模板办法 设计模式，会匹配每一个插件，直到找到 divide 插件

public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        // 获取插件名称
        String pluginName = named();
        // 从缓存中获取插件信息
        final PluginData pluginData = BaseDataCache.getInstance().obtainPluginData(pluginName);
        if (pluginData != null && pluginData.getEnabled()) {
             // 从缓存中依据插件名称获取对应选择器列表
            final Collection<SelectorData> selectors = BaseDataCache.getInstance().obtainSelectorData(pluginName);
            if (CollectionUtils.isEmpty(selectors)) {return handleSelectorIsNull(pluginName, exchange, chain);
            }
             // 匹配选择器
            final SelectorData selectorData = matchSelector(exchange, selectors);
            if (Objects.isNull(selectorData)) {return handleSelectorIsNull(pluginName, exchange, chain);
            }
            selectorLog(selectorData, pluginName);
            // 从缓存中获取选择器对应的规定列表
            final List<RuleData> rules = BaseDataCache.getInstance().obtainRuleData(selectorData.getId());
            if (CollectionUtils.isEmpty(rules)) {return handleRuleIsNull(pluginName, exchange, chain);
            }
            RuleData rule;
            // 不太明确
            if (selectorData.getType() == SelectorTypeEnum.FULL_FLOW.getCode()) {
                //get last
                rule = rules.get(rules.size() - 1);
            } else {rule = matchRule(exchange, rules);
            }
            if (Objects.isNull(rule)) {return handleRuleIsNull(pluginName, exchange, chain);
            }
            ruleLog(rule, pluginName);
            // 调用具体执行办法
            return doExecute(exchange, chain, selectorData, rule);
        }
        return chain.execute(exchange);
    }

而后就回到了最后的终点，在 DividePlugin.java 的doExecute办法中打的第一个断点

protected Mono<Void> doExecute(final ServerWebExchange exchange, final SoulPluginChain chain, final SelectorData selector, final RuleData rule) {final SoulContext soulContext = exchange.getAttribute(Constants.CONTEXT);
        assert soulContext != null;
        final DivideRuleHandle ruleHandle = GsonUtils.getInstance().fromJson(rule.getHandle(), DivideRuleHandle.class);
        // 获取拜访的服务地址，这是是 8188，8189 两条
        final List<DivideUpstream> upstreamList = UpstreamCacheManager.getInstance().findUpstreamListBySelectorId(selector.getId());
        if (CollectionUtils.isEmpty(upstreamList)) {log.error("divide upstream configuration error：{}", rule.toString());
            Object error = SoulResultWrap.error(SoulResultEnum.CANNOT_FIND_URL.getCode(), SoulResultEnum.CANNOT_FIND_URL.getMsg(), null);
            return WebFluxResultUtils.result(exchange, error);
        }
        final String ip = Objects.requireNonNull(exchange.getRequest().getRemoteAddress()).getAddress().getHostAddress();
       // 依据权重，获取拜访的服务地址
        DivideUpstream divideUpstream = LoadBalanceUtils.selector(upstreamList, ruleHandle.getLoadBalance(), ip);
        if (Objects.isNull(divideUpstream)) {log.error("divide has no upstream");
            Object error = SoulResultWrap.error(SoulResultEnum.CANNOT_FIND_URL.getCode(), SoulResultEnum.CANNOT_FIND_URL.getMsg(), null);
            return WebFluxResultUtils.result(exchange, error);
        }
        // set the http url
        String domain = buildDomain(divideUpstream);
        String realURL = buildRealURL(domain, soulContext, exchange);
        exchange.getAttributes().put(Constants.HTTP_URL, realURL);
        // set the http timeout
        exchange.getAttributes().put(Constants.HTTP_TIME_OUT, ruleHandle.getTimeout());
        exchange.getAttributes().put(Constants.HTTP_RETRY, ruleHandle.getRetry());
        return chain.execute(exchange);
    }

DivideUpstream divideUpstream = LoadBalanceUtils.selector(upstreamList, ruleHandle.getLoadBalance(), ip);办法为依据权重获取拜访的服务地址，一路断点，最终找到 RandomLoadBalance 类

public DivideUpstream doSelect(final List<DivideUpstream> upstreamList, final String ip) {
        // 计算总权重
        int totalWeight = calculateTotalWeight(upstreamList);
        // 判断是否平均分配
        boolean sameWeight = isAllUpStreamSameWeight(upstreamList);
        if (totalWeight > 0 && !sameWeight) {
            // 获取具体的拜访地址信息
            return random(totalWeight, upstreamList);
        }
        // If the weights are the same or the weights are 0 then random
        return random(upstreamList);
    }
// 依据权重获取具体的拜访地址信息
private DivideUpstream random(final int totalWeight, final List<DivideUpstream> upstreamList) {
        // If the weights are not the same and the weights are greater than 0, then random by the total number of weights
        int offset = RANDOM.nextInt(totalWeight);
        // Determine which segment the random value falls on
        for (DivideUpstream divideUpstream : upstreamList) {offset -= getWeight(divideUpstream);
            if (offset < 0) {return divideUpstream;}
        }
        return upstreamList.get(0);
    }

至此，devide 插件负载平衡权重剖析告一段落。

正文完
Soul
发表至： Soul
2021-01-19
 0
关于Soul:Soul源码阅读05springcloud插件初体验
关于Soul:Soul源码阅读09数据同步之nacos
关于Soul:Soul源码阅读06数据同步之websocket
关于Soul:Soul源码阅读08数据同步之http长轮询