Midnight

Soul网关中的Sofa插件执行原理（一）

01 Feb 2021   |   Soul  |  

在之前的文章中，我们体验过了Sofa插件执行流程，本篇文章是通过跟踪源码的方式来理解其中的执行原理。

在Soul网关中，Sofa插件负责将http协议转换成sofa协议，涉及到的插件有：BodyParamPlugin，SofaPlugin和SofaResponsePlugin。

  • BodyParamPlugin：负责将请求的json放到exchange属性中；
  • SofaPlugin：使用Sofa进行请求的泛化调用并返回响应结果；
  • SofaResponsePlugin：包装响应结果。

BodyParamPlugin插件

该插件在执行链路中是先执行的，负责处理请求类型，比如带有参数的application/json，不带参数的查询请求。

//org.dromara.soul.plugin.sofa.param.BodyParamPlugin#execute  
public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
		//省略了其他代码
        if (Objects.nonNull(soulContext) && RpcTypeEnum.SOFA.getName().equals(soulContext.getRpcType())) {

            //处理json
            if (MediaType.APPLICATION_JSON.isCompatibleWith(mediaType)) {
                return body(exchange, serverRequest, chain);
            }
            //处理x-www-form-urlencoded
            if (MediaType.APPLICATION_FORM_URLENCODED.isCompatibleWith(mediaType)) {
                return formData(exchange, serverRequest, chain);
            }
            //处理查询
            return query(exchange, serverRequest, chain);
        }
        return chain.execute(exchange);
    }

SofaPlugin插件

完成SofaDubbo插件的核心处理逻辑：检查元数据 -->检查参数类型-->泛化调用。

//org.dromara.soul.plugin.sofa.SofaPlugin#doExecute
@Override
    protected Mono<Void> doExecute(final ServerWebExchange exchange, final SoulPluginChain chain, final SelectorData selector, final RuleData rule) {
		//省略了其他代码
        //检查元数据
        if (!checkMetaData(metaData)) {
			//省略了其他代码
        }
        //检查参数类型
        if (StringUtils.isNoneBlank(metaData.getParameterTypes()) && StringUtils.isBlank(body)) {
           //省略了其他代码
        }
        //泛化调用
        final Mono<Object> result = sofaProxyService.genericInvoker(body, metaData, exchange);
        return result.then(chain.execute(exchange));
    }

genericInvoker()方法中参数分别是请求参数body，有关服务信息的metaData，包含web信息的exchange。这里面的主要操作是：

  • 根据请求路径从缓存获取服务配置信息；
  • 获取代理对象；
  • 请求参数转化为sofa泛化参数；
  • sofa真正的泛化调用；
  • 返回结果。
public Mono<Object> genericInvoker(final String body, final MetaData metaData, final ServerWebExchange exchange) throws SoulException {
    
    	//获取服务信息
        ConsumerConfig<GenericService> reference = ApplicationConfigCache.getInstance().get(metaData.getPath());
		//获取代理对象
        GenericService genericService = reference.refer();
   		 //请求参数转化为泛化调用的参数
        Pair<String[], Object[]> pair;
        if (null == body || "".equals(body) || "{}".equals(body) || "null".equals(body)) {
            pair = new ImmutablePair<>(new String[]{}, new Object[]{});
        } else {
            pair = sofaParamResolveService.buildParameter(body, metaData.getParameterTypes());
        }
    	//异步返回结果
        CompletableFuture<Object> future = new CompletableFuture<>();
    	//响应回调
        RpcInvokeContext.getContext().setResponseCallback(new SofaResponseCallback<Object>() {
            @Override
            public void onAppResponse(final Object o, final String s, final RequestBase requestBase) {
                future.complete(o);
            }

            @Override
            public void onAppException(final Throwable throwable, final String s, final RequestBase requestBase) {
                future.completeExceptionally(throwable);
            }

            @Override
            public void onSofaException(final SofaRpcException e, final String s, final RequestBase requestBase) {
                future.completeExceptionally(e);
            }
        });
    
    //真正的泛化调用
        genericService.$invoke(metaData.getMethodName(), pair.getLeft(), pair.getRight());
    //返回结果    
    return Mono.fromFuture(future.thenApply(ret -> {
            if (Objects.isNull(ret)) {
                ret = Constants.SOFA_RPC_RESULT_EMPTY;
            }
            exchange.getAttributes().put(Constants.SOFA_RPC_RESULT, ret);
            exchange.getAttributes().put(Constants.CLIENT_RESPONSE_RESULT_TYPE, ResultEnum.SUCCESS.getName());
            return ret;
        })).onErrorMap(SoulException::new);
    }

真正的sofa泛化调用是genericService.$invoke()。到这里，SofaPlugin插件的主要工作就完了，后面就是返回结果。

SofaResponsePlugin插件

这个插件就是对结果再一次包装，处理错误信息和成功的结果信息。

//org.dromara.soul.plugin.sofa.response.SofaResponsePlugin#execute   
    @Override
    public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        return chain.execute(exchange).then(Mono.defer(() -> {
            final Object result = exchange.getAttribute(Constants.SOFA_RPC_RESULT);
            //处理错误的信息
            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));
            return WebFluxResultUtils.result(exchange, success);
        }));
    }

至此，就跟踪完了Soul网关中对Sofa插件处理的核心操作：接入sofa服务，将http访问协议转化为sofa协议，通过sofa的泛化调用获取真正的接口服务信息。

Soul网关中的Apache Dubbo插件执行原理（二）

30 Jan 2021   |   Soul  |  

在上一篇文章中，我们通过跟踪源码的方式理解了Apache Dubbo插件的执行原理，将发起的http请求转化为dubbo的泛化调用，但是但是有个关键的地方没有展开讲：就是服务的配置信息是怎么来的？以及代理对象是怎么得到的？

public Mono<Object> genericInvoker(final String body, final MetaData metaData, final ServerWebExchange exchange) throws SoulException {
        // 省略了其他代码

    	//获取服务配置
        ReferenceConfig<GenericService> reference = ApplicationConfigCache.getInstance().get(metaData.getPath());
		//获得代理对象
        GenericService genericService = reference.get();

    }

这篇文章，我们就来解决这个问题。先说结论：服务配置信息的来源是soul-admin同步到网关。

整个流程涉及到的类和方法如下：

soul-admin端：

  • WebsocketCollector:onMessage()；
  • SyncDataServiceImpl:syncAll();
  • MetaDataService:syncData();

soul-bootstrap网关端：

  • SoulWebsocketClient: onMessage(),handleResult();
  • WebsocketDataHandler:executor();
  • AbstractDataHandler:handle();
  • MetaDataHandler:doRefresh();
  • ApacheDubboMetaDataSubscriber:onSubscribe();
  • ApplicationConfigCache:initRef(),build();

在本文的测试中，soul-admin端和soul-bootstrap网关端之间的数据同步是通过websocket进行。所以在网关启动的时候会进行数据同步操作。

在soul-bootstrap网关端前面几个类主要处理同步的数据类型，最终流转到ApplicationConfigCache，在这里面进行构建服务build()。

public ReferenceConfig<GenericService> build(final MetaData metaData) {
        ReferenceConfig<GenericService> reference = new ReferenceConfig<>(); //构建服务
        reference.setGeneric(true);
        reference.setApplication(applicationConfig);
        reference.setRegistry(registryConfig); //注册地址，比如zk
        reference.setInterface(metaData.getServiceName());
        reference.setProtocol("dubbo"); //dubbo协议
        String rpcExt = metaData.getRpcExt();
        DubboParamExtInfo dubboParamExtInfo = GsonUtils.getInstance().fromJson(rpcExt, DubboParamExtInfo.class);
        if (Objects.nonNull(dubboParamExtInfo)) {
            if (StringUtils.isNoneBlank(dubboParamExtInfo.getVersion())) {
                reference.setVersion(dubboParamExtInfo.getVersion()); //版本
            }
            if (StringUtils.isNoneBlank(dubboParamExtInfo.getGroup())) {
                reference.setGroup(dubboParamExtInfo.getGroup());
            }
            if (StringUtils.isNoneBlank(dubboParamExtInfo.getLoadbalance())) {
                final String loadBalance = dubboParamExtInfo.getLoadbalance();
                reference.setLoadbalance(buildLoadBalanceName(loadBalance)); //负载均衡
            }
            if (StringUtils.isNoneBlank(dubboParamExtInfo.getUrl())) {
                reference.setUrl(dubboParamExtInfo.getUrl());
            }
            Optional.ofNullable(dubboParamExtInfo.getTimeout()).ifPresent(reference::setTimeout);
            Optional.ofNullable(dubboParamExtInfo.getRetries()).ifPresent(reference::setRetries);
        }
        try {
            Object obj = reference.get(); //从注册中心中获得代理
            if (obj != null) {
                log.info("init apache dubbo reference success there meteData is :{}", metaData.toString());
                cache.put(metaData.getPath(), reference); //放到cache中
            }
        } catch (Exception e) {
            log.error("init apache dubbo reference ex:{}", e.getMessage());
        }
        return reference;
    }

一个reference的信息如下：包括协议，服务接口，超时时间，负载均衡算法，重试次数。

<dubbo:reference protocol="dubbo" prefix="dubbo.reference.org.dromara.soul.examples.dubbo.api.service.DubboMultiParamService" uniqueServiceName="org.dromara.soul.examples.dubbo.api.service.DubboMultiParamService" interface="org.dromara.soul.examples.dubbo.api.service.DubboMultiParamService" generic="true" generic="true" sticky="false" loadbalance="random" retries="2" timeout="10000" id="org.dromara.soul.examples.dubbo.api.service.DubboMultiParamService" valid="true" />

一个代理对象信息如下：接口信息，注册中心地址等等其他详细信息。

invoker :interface org.apache.dubbo.rpc.service.GenericService -> zookeeper://localhost:2181/org.apache.dubbo.registry.RegistryService?anyhost=true&application=soul_proxy&check=false&deprecated=false&dubbo=2.0.2&dynamic=true&generic=true&interface=org.dromara.soul.examples.dubbo.api.service.DubboMultiParamService&loadbalance=random&methods=batchSave,findByArrayIdsAndName,findByListId,batchSaveAndNameAndId,findByIdsAndName,saveComplexBeanTestAndName,saveComplexBeanTest,findByStringArray&pid=23576&protocol=dubbo&register.ip=192.168.236.60&release=2.7.5&remote.application=test-dubbo-service&retries=2&side=consumer&sticky=false&timeout=10000&timestamp=1612011330405,directory: org.apache.dubbo.registry.integration.RegistryDirectory@406951e0

在同步过程中，将所有的元数据信息及注册中心的服务放到了cache中，在使用的时候，就会到这个cache中去拿。就是在文章开始的地方提到的泛化调用。

//org.dromara.soul.plugin.apache.dubbo.proxy.ApacheDubboProxyService#genericInvoker
public Mono<Object> genericInvoker(final String body, final MetaData metaData, final ServerWebExchange exchange) throws SoulException {
        // 省略了其他代码

    	//从cache中获取服务配置
        ReferenceConfig<GenericService> reference = ApplicationConfigCache.getInstance().get(metaData.getPath());
		//获得代理对象，这个对象就是在同步过程中保存的代理对象
        GenericService genericService = reference.get();

    }

小结：本文再次梳理了Soul网关中Apache Dubbo插件里面的服务是如何被加载的，包括服务的配置信息和 代理对象的生成。

Soul网关中的Sofa插件执行原理（一）

29 Jan 2021   |   Soul  |  

在之前的文章中，我们体验过了Sofa插件执行流程，本篇文章是通过跟踪源码的方式来理解其中的执行原理。

在Soul网关中，Sofa插件负责将http协议转换成sofa协议，涉及到的插件有：BodyParamPlugin，SofaPlugin和SofaResponsePlugin。

  • BodyParamPlugin：负责将请求的json放到exchange属性中；
  • SofaPlugin：使用Sofa进行请求的泛化调用并返回响应结果；
  • SofaResponsePlugin：包装响应结果。

BodyParamPlugin插件

该插件在执行链路中是先执行的，负责处理请求类型，比如带有参数的application/json，不带参数的查询请求。

//org.dromara.soul.plugin.sofa.param.BodyParamPlugin#execute  
public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
		//省略了其他代码
        if (Objects.nonNull(soulContext) && RpcTypeEnum.SOFA.getName().equals(soulContext.getRpcType())) {

            //处理json
            if (MediaType.APPLICATION_JSON.isCompatibleWith(mediaType)) {
                return body(exchange, serverRequest, chain);
            }
            //处理x-www-form-urlencoded
            if (MediaType.APPLICATION_FORM_URLENCODED.isCompatibleWith(mediaType)) {
                return formData(exchange, serverRequest, chain);
            }
            //处理查询
            return query(exchange, serverRequest, chain);
        }
        return chain.execute(exchange);
    }

SofaPlugin插件

完成SofaDubbo插件的核心处理逻辑：检查元数据 -->检查参数类型-->泛化调用。

//org.dromara.soul.plugin.sofa.SofaPlugin#doExecute
@Override
    protected Mono<Void> doExecute(final ServerWebExchange exchange, final SoulPluginChain chain, final SelectorData selector, final RuleData rule) {
		//省略了其他代码
        //检查元数据
        if (!checkMetaData(metaData)) {
			//省略了其他代码
        }
        //检查参数类型
        if (StringUtils.isNoneBlank(metaData.getParameterTypes()) && StringUtils.isBlank(body)) {
           //省略了其他代码
        }
        //泛化调用
        final Mono<Object> result = sofaProxyService.genericInvoker(body, metaData, exchange);
        return result.then(chain.execute(exchange));
    }

genericInvoker()方法中参数分别是请求参数body，有关服务信息的metaData，包含web信息的exchange。这里面的主要操作是：

  • 根据请求路径从缓存获取服务配置信息；
  • 获取代理对象；
  • 请求参数转化为sofa泛化参数；
  • sofa真正的泛化调用；
  • 返回结果。
public Mono<Object> genericInvoker(final String body, final MetaData metaData, final ServerWebExchange exchange) throws SoulException {
    
    	//获取服务信息
        ConsumerConfig<GenericService> reference = ApplicationConfigCache.getInstance().get(metaData.getPath());
		//获取代理对象
        GenericService genericService = reference.refer();
   		 //请求参数转化为泛化调用的参数
        Pair<String[], Object[]> pair;
        if (null == body || "".equals(body) || "{}".equals(body) || "null".equals(body)) {
            pair = new ImmutablePair<>(new String[]{}, new Object[]{});
        } else {
            pair = sofaParamResolveService.buildParameter(body, metaData.getParameterTypes());
        }
    	//异步返回结果
        CompletableFuture<Object> future = new CompletableFuture<>();
    	//响应回调
        RpcInvokeContext.getContext().setResponseCallback(new SofaResponseCallback<Object>() {
            @Override
            public void onAppResponse(final Object o, final String s, final RequestBase requestBase) {
                future.complete(o);
            }

            @Override
            public void onAppException(final Throwable throwable, final String s, final RequestBase requestBase) {
                future.completeExceptionally(throwable);
            }

            @Override
            public void onSofaException(final SofaRpcException e, final String s, final RequestBase requestBase) {
                future.completeExceptionally(e);
            }
        });
    
    //真正的泛化调用
        genericService.$invoke(metaData.getMethodName(), pair.getLeft(), pair.getRight());
    //返回结果    
    return Mono.fromFuture(future.thenApply(ret -> {
            if (Objects.isNull(ret)) {
                ret = Constants.SOFA_RPC_RESULT_EMPTY;
            }
            exchange.getAttributes().put(Constants.SOFA_RPC_RESULT, ret);
            exchange.getAttributes().put(Constants.CLIENT_RESPONSE_RESULT_TYPE, ResultEnum.SUCCESS.getName());
            return ret;
        })).onErrorMap(SoulException::new);
    }

真正的sofa泛化调用是genericService.$invoke()。到这里，SofaPlugin插件的主要工作就完了，后面就是返回结果。

SofaResponsePlugin插件

这个插件就是对结果再一次包装，处理错误信息和成功的结果信息。

//org.dromara.soul.plugin.sofa.response.SofaResponsePlugin#execute   
    @Override
    public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        return chain.execute(exchange).then(Mono.defer(() -> {
            final Object result = exchange.getAttribute(Constants.SOFA_RPC_RESULT);
            //处理错误的信息
            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));
            return WebFluxResultUtils.result(exchange, success);
        }));
    }

至此，就跟踪完了Soul网关中对Sofa插件处理的核心操作：接入sofa服务，将http访问协议转化为sofa协议，通过sofa的泛化调用获取真正的接口服务信息。