Soul网关源码分析-8期
文章目录
Soul 网关 HTTP 服务探活机制
准备工作
仅启动 soul-admin 与 soul-bootstrap, soul-admin 中有之前测试 Http 转发时注册的服务路径及服务元数据.
在这种情况下, 看看如果没启动相应服务节点的情况会发生什么.
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-sukaHeAm-1611148987358)(images/image-20210120194018195.png)]
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Zc1CABJU-1611148987361)(images/image-20210120193952242.png)]
从一个知道会失败的请求开始
尝试调用一个 http路径 http://localhost:9195/http/test/findByUserId?userId=1, 返回值如下:
{
"code": -106,
"message": "未能找到合适的调用url,请检查你的配置!"
}
继续看看打印的日志信息:
2021-01-20 19:34:42.718 INFO 3929 --- [-work-threads-1] o.d.soul.plugin.base.AbstractSoulPlugin : divide selector success match , selector name :/http
2021-01-20 19:34:42.719 INFO 3929 --- [-work-threads-1] o.d.soul.plugin.base.AbstractSoulPlugin : divide rule success match ,rule name :/http/test/**
2021-01-20 19:34:42.723 ERROR 3929 --- [-work-threads-1] o.d.soul.plugin.divide.DividePlugin : divide upstream configuration error：RuleData(id=1349650853992337408, name=/http/test/**, pluginName=divide, selectorId=1349650852775989248, matchMode=0, sort=1, enabled=true, loged=true, handle={"requestVolumeThreshold":"0","errorThresholdPercentage":"0","maxConcurrentRequests":"0","sleepWindowInMilliseconds":"0","loadBalance":"roundRobin","timeout":3000,"retry":"0"}, conditionDataList=[ConditionData(paramType=uri, operator=match, paramName=/, paramValue=/http/test/**)])
从日志可以看出 selector 匹配成功, 因为请求的 /http 与选择器中的相同. rule 匹配也成功, 匹配到 /http/test/**.
最值得注意的是第三行Error级别日志, upstream 相关的错误, 根据日志定位到代码 (仅保留分析代码):
public class DividePlugin extends AbstractSoulPlugin {
protected Mono<Void> doExecute(final ServerWebExchange exchange, final SoulPluginChain chain, final SelectorData selector, final RuleData rule) {
// ...
final List<DivideUpstream> upstreamList = UpstreamCacheManager.getInstance().findUpstreamListBySelectorId(selector.getId());
if (CollectionUtils.isEmpty(upstreamList)) {
LOGGER.error("divide upstream configuration error：{}", rule.toString());
Object error = SoulResultWarp.error(SoulResultEnum.CANNOT_FIND_URL.getCode(), SoulResultEnum.CANNOT_FIND_URL.getMsg(), null);
return WebFluxResultUtils.result(exchange, error);
}
// ..
return chain.execute(exchange);
}
}
这里可以得知出错在 upstreamList 为空, 并且可以看得出, 是通过一个缓存管理器, 传入选择器 id 得到的这个服务集群节点列表.
UpstreamCacheManager 节点更新
那么我们找到了要研究的重点, 这个节点缓存数据怎么更新的呢? 扒扒看这个 UpstreamCacheManager ()
public final class UpstreamCacheManager {
private static final Map<String, List<DivideUpstream>> UPSTREAM_MAP = Maps.newConcurrentMap();
public List<DivideUpstream> findUpstreamListBySelectorId(final String selectorId) {
return UPSTREAM_MAP.get(selectorId);
}
}
通过这段代码可以得知, 维护了一个 UPSTREAM_MAP 做内存缓存. 它的数据来源见下面的代码:
public final class UpstreamCacheManager {
private UpstreamCacheManager() {
boolean check = Boolean.parseBoolean(System.getProperty("soul.upstream.check", "false"));
if (check) {
// 开启定时器
new ScheduledThreadPoolExecutor(1, SoulThreadFactory.create("scheduled-upstream-task", false))
.scheduleWithFixedDelay(this::scheduled, 30, Integer.parseInt(System.getProperty("soul.upstream.scheduledTime", "30")), TimeUnit.SECONDS);
}
}
// 若 UPSTREAM_MAP 中数据不为空, 遍历并检查, 再进行更新or删除
private void scheduled() {
if (UPSTREAM_MAP.size() > 0) {
UPSTREAM_MAP.forEach((k, v) -> {
List<DivideUpstream> result = check(v);
if (result.size() > 0) {
UPSTREAM_MAP.put(k, result);
} else {
UPSTREAM_MAP.remove(k);
}
});
}
}
private List<DivideUpstream> check(final List<DivideUpstream> upstreamList) {
List<DivideUpstream> resultList = Lists.newArrayListWithCapacity(upstreamList.size());
for (DivideUpstream divideUpstream : upstreamList) {
// Http请求服务, 探活
final boolean pass = UpstreamCheckUtils.checkUrl(divideUpstream.getUpstreamUrl());
if (pass) {
resultList.add(divideUpstream);
} else {
log.error("check the url={} is fail ", divideUpstream.getUpstreamUrl());
}
}
return resultList;
}
}
这块代码表明, UpstreamCacheManager 初始化构造器后, 开启定时器定时检测 UPSTREAM_MAP 中服务资源的有效性, 如果检测到服务就更新缓存, 未检测到就删掉对应节点的缓存.
进入它的探活调用 UpstreamCheckUtils.checkUrl() 看看:
public class UpstreamCheckUtils {
public static boolean checkUrl(final String url) {
if (StringUtils.isBlank(url)) {
return false;
}
// 检测IP合法性
if (checkIP(url)) {
String[] hostPort;
if (url.startsWith(HTTP)) {
final String[] http = StringUtils.split(url, "\\/\\/");
hostPort = StringUtils.split(http[1], Constants.COLONS);
} else {
hostPort = StringUtils.split(url, Constants.COLONS);
}
// 传入IP与端口
return isHostConnector(hostPort[0], Integer.parseInt(hostPort[1]));
} else {
return isHostReachable(url);
}
}
private static boolean isHostConnector(final String host, final int port) {
try (Socket socket = new Socket()) {
// Socket连接尝试
socket.connect(new InetSocketAddress(host, port));
} catch (IOException e) {
return false;
}
return true;
}
}
这块的方法很清晰, 最终是使用 Socket 对真实服务节点的IP和端口进行连接并返回成功or失败.
这里可以看到对于Soul网关对于 Http 服务的探活还是很重的, 使用 Socket 这种同步IO的方式直接尝试连接, 探活的服务多了很耗费资源.
UpstreamCacheManager 节点新增
不知道大家是否注意到, 刚刚我们分析的这块仅仅是节点缓存的更新与剔除, 它的来源还漏了, 我们找找 UpstreamCacheManager 下的新增缓存信息的方法:
public final class UpstreamCacheManager {
public void submit(final SelectorData selectorData) {
// 将 selectorData 中 handler 字符串转换成服务节点对象
final List<DivideUpstream> upstreamList = GsonUtils.getInstance().fromList(selectorData.getHandle(), DivideUpstream.class);
if (null != upstreamList && upstreamList.size() > 0) {
UPSTREAM_MAP.put(selectorData.getId(), upstreamList);
} else {
// selectorData 节点中的数据如果不存在, 缓存中也删除对应节点数据
UPSTREAM_MAP.remove(selectorData.getId());
}
}
}
submit() 的方法是被怎么调用的呢? 就是实现了 PluginDataHandler 接口并被 CommonPluginDataSubscriber 的 onSelectorSubscribe() 调用:
public class CommonPluginDataSubscriber implements PluginDataSubscriber {
@Override
public void onSelectorSubscribe(final SelectorData selectorData) {
BaseDataCache.getInstance().cacheSelectData(selectorData);
Optional.ofNullable(handlerMap.get(selectorData.getPluginName())).ifPresent(handler -> handler.handlerSelector(selectorData));
}
}
接下来就是与 soul-admin 建立的 webSocket 通信, 传递 Divide 插件元数据相关了, 这里不继续深入原理和昨天我分析的 Soul网关源码分析-6期 中, 针对插件链的插件元数据更新是一个方式, 有兴趣的同学可以翻看一二. (后面也会出个总结篇, 总结下所有同步配置)
一切仍未结束
这里也会发现一个最最重要的问题, Soul网关接收的数据都是由 soul-admin 来的, 接收元数据的方法 submit() 也是接收到后台 selectorData 对象的 handler 属性, 并解析成服务节点信息对象. 那么后台管理系统的这个传入对象, 它的来源是怎么做到服务节点探活及更新的呢 ?
下期我们会接着继续进行 soul-admin 的探活研究…