【Soulソース読解-02】devideプラグイン負荷等化重み分析

11195 ワード

ソウル

ターゲット

examplesの下にあるhttpサービス

を実行

ドキュメントを学習し、divdeプラグインと組み合わせてhttpリクエストsoulゲートウェイを開始し、httpエージェント

を体験する.
httpサービスに関する依存および構成

httpのプロキシプラグイン

を導入soul-bootstrap工程の下のpom.xmlの下で以下の依存性を導入する

 
 
     org.dromara
     soul-spring-boot-starter-plugin-divide
     ${project.version}
 
 
     org.dromara
     soul-spring-boot-starter-plugin-httpclient
     ${project.version}

soulクライアント(SpringBootユーザー用):

を導入soul-examples-http(あなた自身の実際のサービス)pom.xmlに次の依存が追加されました.

    
         org.dromara
         soul-spring-boot-starter-client-springmvc
         ${last.version}

クライアントアクセス構成の追加:

次の構成を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 handle(@NonNull final ServerWebExchange exchange) {
        MetricsTrackerFacade.getInstance().counterInc(MetricsLabelEnum.REQUEST_TOTAL.getName());
        Optional 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 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 execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        //      
        String pluginName = named();
        //          
        final PluginData pluginData = BaseDataCache.getInstance().obtainPluginData(pluginName);
        if (pluginData != null && pluginData.getEnabled()) {
             //                    
            final Collection 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 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 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 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 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 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プラグインの負荷等化重み解析は一段落した.

logstashインクリメンタル同期mysqlデータからkafka実践

【c++】STLのunordered_map用法まとめ