1.8.2.1 基础类
Soul 中充满了各种各样的插件，插件极大地丰富了 Soul 的功能，而且插件支持热插拔，可扩展新极高，Soul 大方部分功能都是基于插件进行实现的。我们先来看一下 Soul 插件的抽象接口 SoulPlugin 的继承关系：
image.png
我们可以看到 SoulPlugin 下又非常多的实现类，我们再看一下 DividePlugin 的继承情况。
image.png
DividePlugin 继承了AbstractSoulPlugin ，AbstractSoulPlugin 再实现了 SoulPlugin 接口。我们先看一下 SoulPlugin 有哪些方法：
image.png
- execute 插件到执行方法，这里传入 Exchange 即各个插件之间进行通信到桥梁， Chain 插件链。
- getOrder 返回各个插件到执行顺序。我们看 SoulConfiguration 这里是Soul 的核心代码，这里配置了对应的WebHandler ，它是所有插件的起点。我们接下来将 webFulx 会讲到。WebHandler 会根据Order 由小到大排列，定义各个插件的处理顺序。
@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);
}
- name 返回插件名。
-
skip 定义是否需要跳过该插件。
我们接着看 AbstractSoulPlugin 的方法。
image.png
我们先来看一下最重要的 Excute规则,这是个 Default 方法，但做完 Selector 和 Rule 的匹配后，我们直接执行 doExecute 方法，这个方法需要各个子类实现。
@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);
}
我们接着看 DividePlugin 做了啥，其主要是向数据交换区塞入了一部分数据。包括最终访问的地址，超时时间，重试次数。
@Override
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);
// 获取下游服务列表
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);
}
最终调用了 chain.execute(exchange) 方法，我们再来看看调用链是怎么实现的。DefaultSoulPluginChain 它是 WebHandler 的子类，实现了 SoulPluginChain 接口，其主要实现了 execute 方法
@Override
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();
});
}
这里使用了 Mono.defer 区别于 Mono.just 是立即执行 just 里的 function ，defer 是延迟加载的，当每次请求进来都会生成一个 DefaultSoulPluginChain ，这里记录了当情处理走到了那个 plugin 。
最后我们看一下 SoulWebHandler ，他是所有WebFlux 请求的处理类。
@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);
}
首先它对所有对插件进行排序，然后插件链，进行链式调度。
网友评论