任务

  1. 运行examples下面的 sofa-rpc服务
  2. 学习文档，结合sofa插件，发起http请求soul网关，体验sofa代理
  3. 遗留任务：网关中 http divide 插件原理的实现，前置知识点：响应式编程，webflux。

sofa 插件初体验

soul-admin & soul-bootstrap

soul-admin 打开对 sofa 插件的支持，soul-bootstrap 中引入对 sofa 插件的支持，重启网关。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
<dependency>
   <groupId>com.alipay.sofa</groupId>
   <artifactId>sofa-rpc-all</artifactId>
   <version>5.7.6</version>
</dependency>
<dependency>
   <groupId>org.apache.curator</groupId>
   <artifactId>curator-client</artifactId>
   <version>4.0.1</version>
</dependency>
<dependency>
   <groupId>org.apache.curator</groupId>
   <artifactId>curator-framework</artifactId>
   <version>4.0.1</version>
</dependency>
<dependency>
   <groupId>org.apache.curator</groupId>
   <artifactId>curator-recipes</artifactId>
   <version>4.0.1</version>
</dependency>
<dependency>
   <groupId>org.dromara</groupId>
   <artifactId>soul-spring-boot-starter-plugin-sofa</artifactId>
   <version>${project.version}</version>
</dependency>

soul-examples-sofa

启动 examples 下的 soul-examples-sofa 示例。请求 /sofa/findAll，成功，结果如下。

image.png

响应式编程

https://spring.io/guides/gs/reactive-rest-service/

divide 插件原理

遗留问题：网关是如何通过 divide 插件进行 http 的代理的？

概述

由于对响应式编程webflux还不够熟，更多是通过猜测。

整体实现流程：（责任链模式）

  • 请求链：DividePlugin doExecute -> AbstractSoulPlugin（SoulPlugin） execute -> DefaultSoulPluginChain（SoulPluginChain） execute -> WebClientPlugin（SoulPlugin） execute。

插件的抽象类 AbstractSoulPlugin 使用了模板方法模式，自定义插件 DividePlugin 继承了抽象类，重点实现 doExecute 方法。DividePlugin 的功能是处理路由到网关的请求。

DividePlugin：

  1. 读取规则。
  2. 从 UpstreamCacheManager 中读取 DivideUpstream 的列表。
  3. 使用工具 LoadBalanceUtils 决定 DivideUpstream 的负载均衡策略。
  4. 构造 ServerWebExchange。
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    protected Mono<Void> doExecute(final ServerWebExchange exchange, final SoulPluginChain chain, final SelectorData selector, final RuleData rule) {
       ...
       final DivideRuleHandle ruleHandle = GsonUtils.getInstance().fromJson(rule.getHandle(), DivideRuleHandle.class);
       final List<DivideUpstream> upstreamList = UpstreamCacheManager.getInstance().findUpstreamListBySelectorId(selector.getId());
       ...
       DivideUpstream divideUpstream = LoadBalanceUtils.selector(upstreamList, ruleHandle.getLoadBalance(), ip);
       ...
       // 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);
    }

DefaultSoulPluginChain：责任链处理所有的 exchange。

1
2
3
4
5
6
7
8
9
10
11
12
13
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();
    });
}

WebClientPlugin：使用 webClient 发送请求。

1
2
3
4
5
6
7
8
9
public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
    ...
    String urlPath = exchange.getAttribute(Constants.HTTP_URL);
    ...
    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);
}

Link