这一节我们来讲一下dubbo请求的泛化调用。
dubbo请求到网关后，会进入到ApacheDubboPlugin中，由于dubbo版本的不同，还有一个AlibabaDubboPlugin。我们先看ApachaDubboPlugin,他们逻辑基本相似。
请求来了后，会进入到Dubbo插件
//org.dromara.soul.plugin.apache.dubbo.ApacheDubboPlugin#doExecute
protected Mono<Void> doExecute(final ServerWebExchange exchange, final SoulPluginChain chain, final SelectorData selector, final RuleData rule) {
//先获取dubbo的参数
String body = exchange.getAttribute(Constants.DUBBO_PARAMS);
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);
}
//如果元数据的参数列表不为空但是我们的请求body为空，则返回异常
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);
}
//进行dubbo的泛化调用并返回结果
final Mono<Object> result = dubboProxyService.genericInvoker(body, metaData, exchange);
return result.then(chain.execute(exchange));
}
我们来看一下泛化调用
//org.dromara.soul.plugin.apache.dubbo.proxy.ApacheDubboProxyService#genericInvoker
public Mono<Object> genericInvoker(final String body, final MetaData metaData, final ServerWebExchange exchange) throws SoulException {
//从header中获取DubboTag信息
String dubboTagRouteFromHttpHeaders = exchange.getRequest().getHeaders().getFirst(Constants.DUBBO_TAG_ROUTE);
//如果DubboTag不为空，则放到RpcContext的附件中，用于dubbo传播中使用
if (StringUtils.isNotBlank(dubboTagRouteFromHttpHeaders)) {
RpcContext.getContext().setAttachment(CommonConstants.TAG_KEY, dubboTagRouteFromHttpHeaders);
}
//获取dubbo泛化配置
ReferenceConfig<GenericService> reference = ApplicationConfigCache.getInstance().get(metaData.getPath());
//如果获取为空，则通过initRef获取
if (Objects.isNull(reference) || StringUtils.isEmpty(reference.getInterface())) {
ApplicationConfigCache.getInstance().invalidate(metaData.getPath());
reference = ApplicationConfigCache.getInstance().initRef(metaData);
}
//通过泛化配置获取到泛化Service
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));
}
重点是如何获取到泛化配置的. 是通过ApplicationConfigCache.getInstance().initRef(metaData);获取到。
我们来看下ApplicationConfigCache
//org.dromara.soul.plugin.apache.dubbo.cache.ApplicationConfigCache
public final class ApplicationConfigCache {
private ApplicationConfig applicationConfig;
private RegistryConfig registryConfig;
}
他有两个dubbo的基本配置，应用配置和注册中心配置，这两个配置是为了构造泛化调用所必须的。他通过init方法注入到类中
//org.dromara.soul.plugin.apache.dubbo.cache.ApplicationConfigCache#init
public void init(final DubboRegisterConfig dubboRegisterConfig) {
if (applicationConfig == null) {
applicationConfig = new ApplicationConfig("soul_proxy");
}
if (registryConfig == null) {
registryConfig = new RegistryConfig();
registryConfig.setProtocol(dubboRegisterConfig.getProtocol());
registryConfig.setId("soul_proxy");
registryConfig.setRegister(false);
//这里是我们从dubbo服务注册的dubbo插件里面的注册中心，从而能够保证我们soul网关代理的dubbo服务的注册中心和我们服务的注册中心使用同一个，从而能够保证服务调用通畅
registryConfig.setAddress(dubboRegisterConfig.getRegister());
Optional.ofNullable(dubboRegisterConfig.getGroup()).ifPresent(registryConfig::setGroup);
}
}
init方法调用时机，就是之前我们讲的PluginDataHandler调用时机调用的。这里用到ApacheDubboPluginDataHandler
//org.dromara.soul.plugin.apache.dubbo.handler.ApacheDubboPluginDataHandler#handlerPlugin
public void handlerPlugin(final PluginData pluginData) {
if (null != pluginData && pluginData.getEnabled()) {
DubboRegisterConfig dubboRegisterConfig = GsonUtils.getInstance().fromJson(pluginData.getConfig(), DubboRegisterConfig.class);
DubboRegisterConfig exist = Singleton.INST.get(DubboRegisterConfig.class);
if (Objects.isNull(dubboRegisterConfig)) {
return;
}
if (Objects.isNull(exist) || !dubboRegisterConfig.equals(exist)) {
// If it is null, initialize it
ApplicationConfigCache.getInstance().init(dubboRegisterConfig);
ApplicationConfigCache.getInstance().invalidateAll();
}
Singleton.INST.single(DubboRegisterConfig.class, dubboRegisterConfig);
}
}
我们看到，通过维护内存的DubboPlugin插件，从而更新dubbo的配置缓存，并刷新ApplicationConfigCache，dubbo配置缓存。
当我们第一次从缓存中获取ReferenceConfig的时候，肯定是空的，所以会走到initRef
//org.dromara.soul.plugin.apache.dubbo.cache.ApplicationConfigCache#initRef
public ReferenceConfig<GenericService> initRef(final MetaData metaData) {
try {
//从缓存获取，不为空直接返回
ReferenceConfig<GenericService> referenceConfig = cache.get(metaData.getPath());
if (StringUtils.isNoneBlank(referenceConfig.getInterface())) {
return referenceConfig;
}
} catch (ExecutionException e) {
log.error("init dubbo ref ex:{}", e.getMessage());
}
//build
return build(metaData);
}
接下来看下build方法
//org.dromara.soul.plugin.apache.dubbo.cache.ApplicationConfigCache#build
public ReferenceConfig<GenericService> build(final MetaData metaData) {
//手动构造ReferenceConfig
ReferenceConfig<GenericService> reference = new ReferenceConfig<>();
reference.setGeneric(true);
//注入applicationConfig
reference.setApplication(applicationConfig);
//注入registryConfig
reference.setRegistry(registryConfig);
//注入serviceName
reference.setInterface(metaData.getServiceName());
reference.setProtocol("dubbo");
//从metaData的rpcExt中获取额外的信息
String rpcExt = metaData.getRpcExt();
DubboParamExtInfo dubboParamExtInfo = GsonUtils.getInstance().fromJson(rpcExt, DubboParamExtInfo.class);
if (Objects.nonNull(dubboParamExtInfo)) {
//设置dubbo的version
if (StringUtils.isNoneBlank(dubboParamExtInfo.getVersion())) {
reference.setVersion(dubboParamExtInfo.getVersion());
}
//设置dubbo的group
if (StringUtils.isNoneBlank(dubboParamExtInfo.getGroup())) {
reference.setGroup(dubboParamExtInfo.getGroup());
}
//设置负载均衡策略
if (StringUtils.isNoneBlank(dubboParamExtInfo.getLoadbalance())) {
final String loadBalance = dubboParamExtInfo.getLoadbalance();
reference.setLoadbalance(buildLoadBalanceName(loadBalance));
}
//设置url
if (StringUtils.isNoneBlank(dubboParamExtInfo.getUrl())) {
reference.setUrl(dubboParamExtInfo.getUrl());
}
Optional.ofNullable(dubboParamExtInfo.getTimeout()).ifPresent(reference::setTimeout);
Optional.ofNullable(dubboParamExtInfo.getRetries()).ifPresent(reference::setRetries);
}
try {
//通过get获取代理的对象，如果能够获取到，put到cache中
Object obj = reference.get();
if (obj != null) {
log.info("init apache dubbo reference success there meteData is :{}", metaData.toString());
cache.put(metaData.getPath(), reference);
}
} catch (Exception e) {
log.error("init apache dubbo reference ex:{}", e.getMessage());
}
return reference;
}
至此，网关如何实现的dubbo调用就讲完了，主要依赖的就是dubbo自己的泛化调用，并通过内存维护dubbo的相关配置，从而构造泛化配置，获取provider代理，并直接调用
网友评论