任务

  • apache dubbo 的底层原理

dubbo 底层原理

核心工程

核心工程：soul-plugin-apache-dubbo

image.png

基本实现原理

ApacheDubboPlugin

  • doExecute：对元数据和 body 内容进行检查
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
protected Mono<Void> doExecute(final ServerWebExchange exchange, final SoulPluginChain chain, final SelectorData selector, final RuleData rule) {
    ...
    //检查元数据
    if (!checkMetaData(metaData)) {
        ...
        return WebFluxResultUtils.result(exchange, error);
    }
    //如果 body，即带参为空，则报 Dubbo must have body param, please enter the JSON format in the body! 错误
    if (StringUtils.isNoneBlank(metaData.getParameterTypes()) && StringUtils.isBlank(body)) {
        ...
        return WebFluxResultUtils.result(exchange, error);
    }
    final Mono<Object> result = dubboProxyService.genericInvoker(body, metaData, exchange);
    return result.then(chain.execute(exchange));
}

ApacheDubboProxyService

  • genericInvoker：通过参数和元数据执行 dubbo 方法
  • 这里的关键是 ReferenceConfig 的配置，可以关注 ApplicationConfigCache 如何缓存数据
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
public Mono<Object> genericInvoker(final String body, final MetaData metaData, final ServerWebExchange exchange) throws SoulException {
    ...
    // 这里是核心，通过元数据的 path 得到 dubbo 的 ReferenceConfig
    ReferenceConfig<GenericService> reference = ApplicationConfigCache.getInstance().get(metaData.getPath());
    ...
    // 获得 GenericService
    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());
    }
    // dubbo 异步调用 method
    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));
}

ApplicationConfigCache

  • 这里用到了 google guava 的 cache 数据变化实现回调的监听器RemovalListener，即可以监听到 cache put/init 时的行为。
  • 因此，这里可以回溯找到 cache init 的行为的触发。

image.png

ApacheDubboPluginDataHandler

  • handlerPlugin：通过 pluginData 执行 ApplicationConfigCache 的 init 方法。
  • pluginData：{"dubbo":{"id":"6","name":"dubbo","config":"{\"register\":\"nacos://localhost:8848\"}","role":1,"enabled":true}}
1
2
3
4
5
6
7
8
9
10
11
12
13
public void handlerPlugin(final PluginData pluginData) {
    if (null != pluginData && pluginData.getEnabled()) {
        // dubbo 的配置
        DubboRegisterConfig dubboRegisterConfig = GsonUtils.getInstance().fromJson(pluginData.getConfig(), DubboRegisterConfig.class);
        ...
        if (Objects.isNull(exist) || !dubboRegisterConfig.equals(exist)) {
            // If it is null, initialize it
            ApplicationConfigCache.getInstance().init(dubboRegisterConfig);
            ApplicationConfigCache.getInstance().invalidateAll();
        }
        ...
    }
}

小结

apache dubbo 的请求逻辑大致是：

  • 网关通过数据同步，将 pluginData 的数据加载到 ApplicationConfigCache 缓存中。
  • 接口请求时，读取 ApplicationConfigCache 缓存中，读取 dubbo 的 service，远程调用真实的 dubbo 服务。