Soul網關的數據註冊和同步數據流探究

Soul網關的Http/SpringMvc 數據註冊和同步數據流探究

Soul-Admin端數據的探究

首先啓動客戶端項目soul-examples-http的過程當中看到控制檯會輸出和

能夠很明顯的看到這段信息就是咱們註解了@SoulSpringMvcClient的接口信息，那麼這個操做是在哪裏產生的了。
全局搜索了register success 字樣。發如今soul-examples-http依賴的子項目soul-client-springmvc中有對應的SpringMvcClientBeanPostProcessor來進行註冊的代碼，這個SpringMvcClientBeanPostProcessor實現了Spring的BeanPostProcessor接口，若是咱們想在Spring容器中完成bean實例化、配置以及其餘初始化方法先後要添加一些本身邏輯處理。咱們須要定義一個或多個BeanPostProcessor接口實現類，而後註冊到Spring IoC容器中。所以這裏就是soul的客戶端將數據註冊到soul-admin的入口，
因而我在註冊這裏進行了斷點調試，看看他究竟是在那個接口註冊從而將數據傳遞到soul-admin項目和網關，

能夠看到http的項目是在控制檯項目的http://localhost:9095/soul-client/springmvc-register 接口進行了註冊。
隨後咱們轉到這個接口

/**
     * Register spring cloud string.
     *
     * @param springCloudRegisterDTO the spring cloud register dto
     * @return the string
     */
    @PostMapping("/springcloud-register")
    public String registerSpringCloud(@RequestBody final SpringCloudRegisterDTO springCloudRegisterDTO) {
        return soulClientRegisterService.registerSpringCloud(springCloudRegisterDTO);
    }

深刻到service層能夠看到是應用的數據利用到上一節的Spring的事件處理機制來實現了

隨後斷點進入到實現了Spring的事件監聽接口ApplicationListener的事件分發類DataChangedEventDispatcher

能夠看到，我啓動的時候由於沒有增長插件，因此此時是進入了SELECTOR斷點中，因爲此時我選擇的zookeepr做爲數據同步的組件。我找到了ZookeeperDataChangedListener中進行斷點 ，發現這裏數據同步是將數據放入到zookeeper

private void createZkNode(final String path) {
        if (!zkClient.exists(path)) {
            zkClient.createPersistent(path, true);
        }
    }
    
    /**
     * create or update zookeeper node.
     * @param path node path
     * @param data node data 
     */
    private void upsertZkNode(final String path, final Object data) {
        if (!zkClient.exists(path)) {
            zkClient.createPersistent(path, true);
        }
        zkClient.writeData(path, data);
    }
    
    private void deleteZkPath(final String path) {
        if (zkClient.exists(path)) {
            zkClient.delete(path);
        }
    }
    
    private void deleteZkPathRecursive(final String path) { 
        if (zkClient.exists(path)) {
            zkClient.deleteRecursive(path);
        }
    }

zookeeper內寫入節點和更新數據的流程如上

Soul-Boostrap端數據的探究

打開soul-boostrap能夠看到。類不多，只有兩個。一個SoulNettyWebServerFactory和HealthFilter。HealthFilter是用來作服務健康檢查的。而SoulNettyWebServerFactory就是spring webflux應用的一個響應式Server工廠類。具體的能夠去看https://www.jianshu.com/p/ada196969995 這篇文章
可是此時咱們仍是沒有達到咱們想要的請求轉發的東西。去pom文件中找到了核心的網關的項目soul-spring-boot-starter-gateway可是發現其中一個類都沒有，可是其中依賴了soul-web模塊，所以對soul-web模塊進行了解
首先能夠看到配置類SoulConfiguration,

/**
     * Init SoulWebHandler.
     *
     * @param plugins this plugins is All impl SoulPlugin.
     * @return {@linkplain SoulWebHandler}
     */
    @Bean("webHandler")
    public SoulWebHandler soulWebHandler(final ObjectProvider<List<SoulPlugin>> plugins) {
        List<SoulPlugin> pluginList = plugins.getIfAvailable(Collections::emptyList);
        final List<SoulPlugin> soulPlugins = pluginList.stream()
                .sorted(Comparator.comparingInt(SoulPlugin::getOrder)).collect(Collectors.toList());
        soulPlugins.forEach(soulPlugin -> log.info("load plugin:[{}] [{}]", soulPlugin.named(), soulPlugin.getClass().getName()));
        return new SoulWebHandler(soulPlugins);
    }

能夠看到這裏,加載了一個webhandler的處理器，處理器的主要內容

public SoulWebHandler(final List<SoulPlugin> plugins) {
        this.plugins = plugins;
        String schedulerType = System.getProperty("soul.scheduler.type", "fixed");
        if (Objects.equals(schedulerType, "fixed")) {
            int threads = Integer.parseInt(System.getProperty(
                    "soul.work.threads", "" + Math.max((Runtime.getRuntime().availableProcessors() << 1) + 1, 16)));
            scheduler = Schedulers.newParallel("soul-work-threads", threads);
        } else {
            scheduler = Schedulers.elastic();
        }
    }
    
    /**
     * Handle the web server exchange.
     *
     * @param exchange the current server exchange
     * @return {@code Mono<Void>} to indicate when request handling is complete
     */
    @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)));
    }

主要內容是利用責任鏈對請求的線程數進行處理。handle方法用來處理請求
根據soul-example-springmvc的註解進行請求。能夠在抽象的基礎AbstractSoulPlugin接受到這個請求

@Override
    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);
    }

這裏有很明顯的規則和選擇器比較的相關邏輯，用來判斷當前的請求是否位於網關代理的請求中。可是在這裏與上面對應的是，我如何取到上文已經設置的選擇器數據，這個留到後面繼續去探究
緊接着斷點來到了具體的業務的請求插件WebClientPlugin執行後面的業務請求

@Override
    public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        final SoulContext soulContext = exchange.getAttribute(Constants.CONTEXT);
        assert soulContext != null;
        String urlPath = exchange.getAttribute(Constants.HTTP_URL);
        if (StringUtils.isEmpty(urlPath)) {
            Object error = SoulResultWrap.error(SoulResultEnum.CANNOT_FIND_URL.getCode(), SoulResultEnum.CANNOT_FIND_URL.getMsg(), null);
            return WebFluxResultUtils.result(exchange, error);
        }
        long timeout = (long) Optional.ofNullable(exchange.getAttribute(Constants.HTTP_TIME_OUT)).orElse(3000L);
        int retryTimes = (int) Optional.ofNullable(exchange.getAttribute(Constants.HTTP_RETRY)).orElse(0);
        log.info("The request urlPath is {}, retryTimes is {}", urlPath, retryTimes);
        HttpMethod method = HttpMethod.valueOf(exchange.getRequest().getMethodValue());
        WebClient.RequestBodySpec requestBodySpec = webClient.method(method).uri(urlPath);
        return handleRequestBody(requestBodySpec, exchange, timeout, retryTimes, chain);
    }

從如上能夠看到，網關在代理請求這一塊的邏輯

問題

本文還剩下未解決的問題主要是

• 如何從abstractSoulPlugin執行完以後到WebClientPlugin的相同方法，是責任鏈模式仍是其餘的加載過程
• abstractSoulPlugin是如何加載註冊或修改後的選擇器等數據
• plugin 中的執行方法是如何獲取到ServerWebExchange的相關請求數據

參考文章 https://blog.csdn.net/u010084384/article/details/113010594

歡迎搜索關注本人與朋友共同開發的微信面經小程序【大廠面試助手】和公衆號【微瞰技術】，以及總結的分類面試題https://github.com/zhendiao/JavaInterview