Soul源码阅读 alibaba-dubbo&apache-du

环境搭建

dubbo运行插件查看 Soul源码阅读 体验dubbo代理【第三天】

执行流程

GlobalPlugin -> BodyParamPlugin [org.dromara.soul.plugin.alibaba.dubbo.param] -> BodyParamPlugin [org.dromara.soul.plugin.sofa.param] 经过这一层但是是无效的 -> AbstractSoulPlugin-> AlibabaDubboPlugin ->DubboResponsePlugin -> 结束

alibab-dubbo 代码解析

BodyParamPlugin [org.dromara.soul.plugin.alibaba.dubbo.param]

@Override
    // dubbo参数解析
    public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        // 请求信息
        final ServerHttpRequest request = exchange.getRequest();
        // Soul上下文
        final SoulContext soulContext = exchange.getAttribute(Constants.CONTEXT);
        // 如果是Dubbo的参数进行解析
        if (Objects.nonNull(soulContext) && RpcTypeEnum.DUBBO.getName().equals(soulContext.getRpcType())) {
            // 获取请求类型
            MediaType mediaType = request.getHeaders().getContentType();
            // webflux请求
            ServerRequest serverRequest = ServerRequest.create(exchange, messageReaders);
            // json请求
            if (MediaType.APPLICATION_JSON.isCompatibleWith(mediaType)) {
                return body(exchange, serverRequest, chain);
            }
            // 表单请求
            if (MediaType.APPLICATION_FORM_URLENCODED.isCompatibleWith(mediaType)) {
                return formData(exchange, serverRequest, chain);
            }
            // 塞入转换后的参数信息
            return query(exchange, serverRequest, chain);
        }
        return chain.execute(exchange);
    }

AlibabaDubboPlugin

@Override
    // 执行dubbo调用
    protected Mono<Void> doExecute(final ServerWebExchange exchange, final SoulPluginChain chain, final SelectorData selector, final RuleData rule) {
        // 获取之前解析的参数信息
        String body = exchange.getAttribute(Constants.DUBBO_PARAMS);
        // Soul上下文
        SoulContext soulContext = exchange.getAttribute(Constants.CONTEXT);
        assert soulContext != null;
        // 获取元数据
        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.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);
    }

AlibabaDubboPlugin使用AlibabaDubboProxyService执行泛化调用

/**
     * Generic invoker object.
     *
     * @param body     the body
     * @param metaData the meta data
     * @return the object
     * @throws SoulException the soul exception
     */
    // 根据参数信息和元数据信息执行泛化调用
    public Object genericInvoker(final String body, final MetaData metaData) throws SoulException {
        // 从缓存中获取引用？
        ReferenceConfig<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);
        }
        // 获取泛化调用服务
        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());
            }
            // 泛化调用
            return genericService.$invoke(metaData.getMethodName(), pair.getLeft(), pair.getRight());
        } catch (GenericException e) {
            log.error("dubbo invoker have exception", e);
            throw new SoulException(e.getExceptionMessage());
        }
    }

DubboResponsePlugin 返回dubbo调用返回的值

/**
     * Process the Web request and (optionally) delegate to the next
     * {@code WebFilter} through the given {@link SoulPluginChain}.
     *
     * @param exchange the current server exchange
     * @param chain    provides a way to delegate to the next filter
     * @return {@code Mono<Void>} to indicate when request processing is complete
     */
    @Override
    // 封装返回数据
    public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        return chain.execute(exchange).then(Mono.defer(() -> {
            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));
                return WebFluxResultUtils.result(exchange, success);
            } catch (SoulException e) {
                return Mono.empty();
            }
        }));
    }

apache dubbo

pom依赖

<!--soul  apache dubbo plugin start-->
         <dependency>
            <groupId>org.dromara</groupId>
            <artifactId>soul-spring-boot-starter-plugin-apache-dubbo</artifactId>
            <version>${project.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.dubbo</groupId>
            <artifactId>dubbo</artifactId>
            <version>2.7.5</version>
        </dependency>

<!-- Dubbo zookeeper registry dependency start -->
        <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>
        <!-- Dubbo zookeeper registry dependency end -->

调用流程

GlobalPlugin -> BodyParamPlugin[org.dromara.soul.plugin.apache.dubbo.param] -> BodyParamPlugin[org.dromara.soul.plugin.sofa.param] -> AbstractSoulPlugin -> ApacheDubboPlugin -> DubboResponsePlugin

ApacheDubboProxyService 主要泛化调用不同

/**
     * Generic invoker object.
     *
     * @param body     the body
     * @param metaData the meta data
     * @param exchange the exchange
     * @return the object
     * @throws SoulException the soul exception
     */
    // apacheDubbo泛化调用
    public Mono<Object> genericInvoker(final String body, final MetaData metaData, final ServerWebExchange exchange) throws SoulException {
        // issue(https://github.com/dromara/soul/issues/471), add dubbo tag route
        // 获取路由信息
        String dubboTagRouteFromHttpHeaders = exchange.getRequest().getHeaders().getFirst(Constants.DUBBO_TAG_ROUTE);
        if (StringUtils.isNotBlank(dubboTagRouteFromHttpHeaders)) {
            RpcContext.getContext().setAttachment(CommonConstants.TAG_KEY, dubboTagRouteFromHttpHeaders);
        }
        // 引用获取
        ReferenceConfig<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);
        }
        // 泛化服务
        GenericService genericService = reference.get();
        Pair<String[], Object[]> pair;
        // 参数封装
        if (ParamCheckUtils.dubboBodyIsEmpty(body)) {
            pair = new ImmutablePair<>(new String[]{}, new Object[]{});
        } else {
            pair = dubboParamResolveService.buildParameter(body, metaData.getParameterTypes());
        }
        //异步调用
        CompletableFuture<Object> future = genericService.$invokeAsync(metaData.getMethodName(), pair.getLeft(), pair.getRight());
        // 异步返回调用链
        return Mono.fromFuture(future.thenApply(ret -> {
            if (Objects.isNull(ret)) {
                ret = Constants.DUBBO_RPC_RESULT_EMPTY;
            }
            exchange.getAttributes().put(Constants.DUBBO_RPC_RESULT, ret);
            exchange.getAttributes().put(Constants.CLIENT_RESPONSE_RESULT_TYPE, ResultEnum.SUCCESS.getName());
            return ret;
        })).onErrorMap(exception -> exception instanceof GenericException ? new SoulException(((GenericException) exception).getExceptionMessage()) : new SoulException(exception));
    }