失眠网 > Soul 网关源码阅读（四）Dubbo请求概览

Soul 网关源码阅读（四）Dubbo请求概览

时间：2024-04-26 23:16:22

简介

本次启动一个dubbo服务示例，初步探索Soul网关源码的Dubbo请求处理流程

示例运行

环境配置

在Soul源码clone下来以后，有一个 soul-example 目录，这个就是示例工程，里面有很多的示例可以运行

可能初始文件夹不被IDEA识别为Java工程，我们需要点击改工程目录下的 pom.xml 文件，然后在右键，在菜单中选择：add as maven project

我们选择运行：soul-examples --> soul-examples-apache-dubbo-service

此次示例需要mysql和zookeeper，我们使用docker启动一下,记得修改 soul-admin 的数据库配置，其他的都不需要懂，猜测zk有默认配置

docker run -dit --name zk -p 2181:2181 zookeepedocker run --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=123456 -d mysql:latest

测试运行

然后运行 Soul-admin、Soul-bootstrap,可以参考：Soul 源码阅读（二）代码初步运行

运行soul-examples-apache-dubbo-service

进入界面：系统管理 --> 插件管理

在插件：dubbo，点击编辑，将状态修改为开启（是个开关图标）

开启成功后，可以在Soul-Bootstrap中看到相关的日志，大致如下：

o.d.s.p.a.d.c.ApplicationConfigCache: init aliaba dubbo reference success there meteData is :MetaDatao.d.s.p.a.d.c.ApplicationConfigCache: init aliaba dubbo reference success there meteData is :MetaDatao.d.s.p.a.d.c.ApplicationConfigCache: init aliaba dubbo reference success there meteData is :MetaData

进入Soul-admin管理界面：插件管理 --> dubbo ，我们可以看到很多的配置，这些都是soul-examples-apache-dubbo-service的配置

随便找个查询的接口：http://localhost:9195/dubbo/findAll

使用浏览器访问得到的结果如下：

{"code": 200,"message": "Access to success!","data": {"name": "hello world Soul Apache, findAll","id": "-1206394682"}}

很好，已经成功跑通示例，接下来，进入源码进行debug

源码分析

基于上篇，我们已经知道了一个HTTP请求的初步处理流程：soul源码阅读3-请求处理概览.md

基于上篇的熟悉度，我们就开始看看Dubbo的处理流程和Http的有什么区别，顺带看一看各个Plugin都干了啥

Plugins的链式处理核心类是：SoulWebHandler，我们就从这里开始打上断点，逐步进入每个Plugin进行查看

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();});}

GlobalPlugin

代码看的不是很懂，猜测有下面的功能：

获取请求头的 upgrade，此时为null，进入逻辑另外的分支的MultiValueMap有点像文件上传之类需要的，后面有时间验证一下功能是修改了SoulContext

这个不是此次分析的目标，到这就跳过

SignPlugin

这个应该是一个认证相关的插件，但在获取插件信息，没有开启，不进入逻辑，直接跳过，没有进入具体执行

这里发现有的能需要进入函数：execute，有的直接进入：doExecute。稍微有点好奇，就看了下原因，大致是直接继承SoulPlugin就直接进入doExecute,而 AbstractSoulPlugin需要进入execute，这里不是重点，就不继续研究这个了，有个大概了解即可

WafPlugin、RateLimiterPlugin、HystrixPlugin、Resilience4JPlugin

上面几个都是获取插件信息，没有开启，不进入逻辑，直接跳过

DividePlugin

这里是true，有点和自己想的不一样，查看管理界面看到divide是开启的，稍微挖一下这个

定位到关键函数：Boolean skip = plugin.skip(exchange)，判断是否可以跳过，我们看下具体的代码：

public Boolean skip(final ServerWebExchange exchange) {final SoulContext soulContext = exchange.getAttribute(Constants.CONTEXT);return !Objects.equals(Objects.requireNonNull(soulContext).getRpcType(), RpcTypeEnum.HTTP.getName());}

好像就是判断类型来返回是否跳过，其他的plugins的判断逻辑，至于这些都是都是怎么来的，今天就不分析了，保留体力，留待下次进行分析

WebClientPulugin、WebsocketPlugin

skip 为 true，直接跳过,跳过的判断的和DividePlugin基本相同，通过类型判断

BodyParamPlugin

对soulContext进行了操作，和GlobalPlugin有一些联动，还看到了MediaType等关键字，感觉很像文件上传之类的，但细节不太清楚，这个也不是本次分析的目的，不要陷入细节，这次放过它，下次有时间再仔细研究

AlibabaDubblePlugin

这个插件是本次的核心，大致干了下面三个事情

进入了并且匹配上了规则，发现这个plugin是dubbo总插件，同样能处理Apache dubbo，也就是他们是相同的或者复用的进入到doExecute函数：获取body，soulContext，查看soulContext发现已经有方法、路径等信息（猜测是前面加的），获得关键的metaData获取请求的结果，并且放到exchange中：Object result = alibabaDubboProxyService.genericInvoker(body, metaData)

代码大致如下：

protected Mono<Void> doExecute(final ServerWebExchange exchange, final SoulPluginChain chain, final SelectorData selector, final RuleData rule) {String body = exchange.getAttribute(Constants.DUBBO_PARAMS);SoulContext soulContext = exchange.getAttribute(Constants.CONTEXT);assert soulContext != null;// dubbo 请求的相关数据MetaData metaData = exchange.getAttribute(Constants.META_DATA);if (!checkMetaData(metaData)) {assert metaData != null;log.error(" path is :{}, meta data have error.... {}", soulContext.getPath(), metaData.toString());exchange.getResponse().setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR);Object error = SoulResultWrap.error(SoulResultEnum.META_DATA_ERROR.getCode(), SoulResultEnum.META_DATA_ERROR.getMsg(), null);return WebFluxResultUtils.result(exchange, error);}if (StringUtils.isNoneBlank(metaData.getParameterTypes()) && StringUtils.isBlank(body)) {exchange.getResponse().setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR);Object error = SoulResultWrap.error(SoulResultEnum.DUBBO_HAVE_BODY_PARAM.getCode(), SoulResultEnum.DUBBO_HAVE_BODY_PARAM.getMsg(), null);return WebFluxResultUtils.result(exchange, error);}// 获取请求结果Object result = alibabaDubboProxyService.genericInvoker(body, metaData);if (Objects.nonNull(result)) {// 将结果放到exchange中exchange.getAttributes().put(Constants.DUBBO_RPC_RESULT, result);} else {exchange.getAttributes().put(Constants.DUBBO_RPC_RESULT, Constants.DUBBO_RPC_RESULT_EMPTY);}exchange.getAttributes().put(Constants.CLIENT_RESPONSE_RESULT_TYPE, ResultEnum.SUCCESS.getName());return chain.execute(exchange);}

继续跟踪查看获取result的那个函数，通过下面的代码可以明显看出确实是rpc调用，并得到结果

public Object genericInvoker(final String body, final MetaData metaData) throws SoulException {// 通过rpc和dubbo的相关知识，这里的reference相当于consumer，metaData.getPath()==/dubbo/findAll// ApplicationConfigCache.getInstance().get 的逻辑感觉比较复杂，先不看了，大体感觉是初始化的时候使用 path 作为了 keyReferenceConfig<GenericService> reference = ApplicationConfigCache.getInstance().get(metaData.getPath());if (Objects.isNull(reference) || StringUtils.isEmpty(reference.getInterface())) {ApplicationConfigCache.getInstance().invalidate(metaData.getPath());reference = ApplicationConfigCache.getInstance().initRef(metaData);}// 这段很像RPC Demo中的获取字节码生成的对象GenericService genericService = reference.get();try {Pair<String[], Object[]> pair;if (ParamCheckUtils.dubboBodyIsEmpty(body)) {pair = new ImmutablePair<>(new String[]{}, new Object[]{});} else {pair = dubboParamResolveService.buildParameter(body, metaData.getParameterTypes());}// 传入方法名、参数名、参数？Dubbo还没用的熟练，暂时这样猜一猜，不影响此次分析的大局return genericService.$invoke(metaData.getMethodName(), pair.getLeft(), pair.getRight());} catch (GenericException e) {log.error("dubbo invoker have exception", e);throw new SoulException(e.getExceptionMessage());}}

对reference.get()，有点好奇，稍微看下 ReferenceConfig

public synchronized T get() {// 判断是否能用if (this.destroyed) {throw new IllegalStateException("Already destroyed!");} else {// 有延迟加载的作用if (this.ref == null) {this.init();}return this.ref;}}

看到它这用法还是挺巧妙的，有一个延迟加载的效果，好像可以按照这个思路去改一改自己的RPC Demo的客户端代理生成，哈哈

MonitorPlugin

获取插件信息，没有开启，不进入逻辑，直接跳过

WebClientResponsePlugin

skip = true，直接跳过

DubboResponsePlugin

看名字就知道是个核心类，我们在then后执行的代码段打个断点（这种需要单独打个端口，后面才能跳进去）

public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {return chain.execute(exchange).then(Mono.defer(() -> {// 从exchange中拿到结果final Object result = exchange.getAttribute(Constants.DUBBO_RPC_RESULT);try {if (Objects.isNull(result)) {Object error = SoulResultWrap.error(SoulResultEnum.SERVICE_RESULT_ERROR.getCode(), SoulResultEnum.SERVICE_RESULT_ERROR.getMsg(), null);return WebFluxResultUtils.result(exchange, error);}Object success = SoulResultWrap.success(SoulResultEnum.SUCCESS.getCode(), SoulResultEnum.SUCCESS.getMsg(), JsonUtils.removeClass(result));// 进入后使用之前熟悉的：exchange.getResponse().writeWith，返回响应给客户端return WebFluxResultUtils.result(exchange, success);} catch (SoulException e) {return Mono.empty();}}));}

执行完plugins链后进入到这个断点：大致是获取结果result，加工得到success

然后WebFluxResultUtils.result(exchange, success)，进入后是：exchange.getResponse().writeWith，返回响应给客户端

到这里，一个Dubbo请求的处理流程大致展示在我们面前了，有了前面几篇的分析基础，还是挺流畅的

总结

此次分析验证了在上篇中的一些猜想，也得到了一些新的东西，请求处理的大致流程图如下：

HttpServerOperations : 明显的netty的请求接收的地方，请求入口ReactorHttpHandlerAdapter ：生成response和requestHttpWebHandlerAdapter ：exchange 的生成FilteringWebHandler : filter 操作SoulWebHandler ：plugins调用

可以参考下上篇分析： Soul 源码阅读（三）HTTP请求处理概览

这篇中我们详细分析各个Dubbo 经过的 Plugin 的相关处理，有些复杂的就没有看了，但也得到我们此次分析想要的结果，脑海中对一个RPC请求如果进行处理，设计那些组件有了一定的了解，基于这些了解也能进行更近一步的分析

我们收获和HTTP请求调用、RPC请求调用、Websocket请求调用是互斥的，通过判断请求的类型来进行选择其中一个，而HTTP相应、RPC响应都会转为HTTP响应，然后返回给客户端，exchange里面估计是绑定了一个socket，然后直接调用即可（Netty网关Demo又可以进行借鉴了）

此次分析有了下面新的疑问：

Websocket的响应的返回时怎样的？是长连接吗？因为没有看到Websocket响应相关的处理类exchange中的请求类型是怎么绑定，如何生成的？限流等插件也是需要进行匹配吗？具体执行逻辑是怎样的？

上面这些问题就留待以后分析了，慢慢来才能可持续发展