22-Dubbo3消费者自动感应决策应用级服务发现原理

22.1 简介

这里要说的内容对Dubbo2迁移到Dubbo3的应用比较有帮助，消费者应用级服务发现做了一些自动决策的逻辑来决定当前消费者是应用级发现还是接口级服务发现，这里与前面说的提供者双注册的原理是对等的，提供者默认同时进行应用级注册和接口级注册，消费者对提供者注册的数据来决定使用应用级发现或者接口级发现。这些都是默认的行为，当然对于消费者来说还可以自定义其他的迁移规则，具体的需要我们详细来看逻辑。

如果说对于迁移过程比较感兴趣可以直接去官网看文档相对来说还是比较清晰:https://dubbo.apache.org/zh/docs/migration/migration-service-discovery/

这里再借官网的图来用用，迁移过程主要如下所示：
第一个图是提供者双注册的图：
在这里插入图片描述

第二个图是消费者订阅决策的图：
在这里插入图片描述

第三个图就是精确到消费者订阅的代码层的逻辑了，消费者服务间调用通过一个Invoker类型对象来进行对象，如下图所示消费者代理对象通过创建一个迁移容错的调用器对象来对应用级或者接口级订阅进行适配如下所示
在这里插入图片描述

第二个图和第三个图是重点要关注的这一个文章的内容主要就是说这里的逻辑。

关于代码位置如果不知道是如何调用到这一块逻辑的可以查看博文《21-Dubbo3消费者引用服务入口》

这里直接将代码位置定位到：RegistryProtocol类型的interceptInvoker方法中：
如下所示：

RegistryProtocol类型的interceptInvoker方法 

  1. protected <T> Invoker<T> interceptInvoker(ClusterInvoker<T> invoker, URL url, URL consumerUrl) {
  2.  //目前存在的扩展类型为RegistryProtocolListener监听器的实现类型MigrationRuleListener 
  3.        List<RegistryProtocolListener> listeners = findRegistryProtocolListeners(url);
  4.        if (CollectionUtils.isEmpty(listeners)) {
  5.            return invoker;
  6.        }
  8.        for (RegistryProtocolListener listener : listeners) {
  9.            listener.onRefer(this, invoker, consumerUrl, url);
  10.        }
  11.        return invoker;
  12.    }

该方法尝试加载所有RegistryProtocolListener定义，这些定义通过与定义的交互来控制调用器的行为，然后使用这些侦听器更改MigrationInvoker的状态和行为。当前可用的监听器是MigrationRuleListener，用于通过动态变化的规则控制迁移行为。

22.2 MigrationRuleListener 类型的onRefer方法

直接来看代码：

  1. @Override
  2.     public void onRefer(RegistryProtocol registryProtocol, ClusterInvoker<?> invoker, URL consumerUrl, URL registryURL) {
  3.        //创建一个对应invoker对象的MigrationRuleHandler类型对象 然后将其存放在缓存Map<MigrationInvoker, MigrationRuleHandler>类型对象handles中
  4.         MigrationRuleHandler<?> migrationRuleHandler = handlers.computeIfAbsent((MigrationInvoker<?>) invoker, _key -> {
  5.             ((MigrationInvoker<?>) invoker).setMigrationRuleListener(this);
  6.             return new MigrationRuleHandler<>((MigrationInvoker<?>) invoker, consumerUrl);
  7.         });
  9.       //迁移规则执行 rule是封装了迁移的配置规则的信息对应类型MigrationRule类型，在初始化对象的时候进行了配置初始化
  10.         migrationRuleHandler.doMigrate(rule);
  11.     }

关于这个igrationRule的文可以直接看官方的文档比较详细：地址迁移规则说明

这个迁移规则是为了更细粒度的迁移决策：
相关配置可以参考下面这个样例：

  1. key: 消费者应用名（必填）
  2. step: 状态名（必填）
  3. threshold: 决策阈值（默认1.0）
  4. proportion: 灰度比例（默认100）
  5. delay: 延迟决策时间（默认0）
  6. force: 强制切换（默认 false）
  7. interfaces: 接口粒度配置（可选）
  8.   - serviceKey: 接口名（接口 + : + 版本号）（必填）
  9.     threshold: 决策阈值
  10.     proportion: 灰度比例
  11.     delay: 延迟决策时间
  12.     force: 强制切换
  13.     step: 状态名（必填）
  14.   - serviceKey: 接口名（接口 + : + 版本号）
  15.     step: 状态名
  16. applications: 应用粒度配置（可选）
  17.   - serviceKey: 应用名（消费的上游应用名）（必填）
  18.     threshold: 决策阈值
  19.     proportion: 灰度比例
  20.     delay: 延迟决策时间
  21.     force: 强制切换
  22.     step: 状态名（必填）

不过为了简单起见暂时先不详细说这个配置细节，我们继续往下看

22.3 迁移规则处理器执行迁移规则MigrationRuleHandler类型的doMigrate方法

22.3.1 迁移规则的模版方法：

MigrationRuleHandler类型的doMigrate方法代码如下：

  1. public synchronized void doMigrate(MigrationRule rule) {
  2.         //默认情况下这个类型是MigrationInvoker
  3.         if (migrationInvoker instanceof ServiceDiscoveryMigrationInvoker) {
  4.             refreshInvoker(MigrationStep.FORCE_APPLICATION, 1.0f, rule);
  5.             return;
  6.         }
  8.       //迁移步骤，MigrationStep 一共有3种枚举情况：FORCE_INTERFACE, APPLICATION_FIRST, FORCE_APPLICATION
  9.         // initial step : APPLICATION_FIRST
  10.         MigrationStep step = MigrationStep.APPLICATION_FIRST;
  11.         float threshold = -1f;
  13.         try {
  14.             //获取配置的类型 默认走APPLICATION_FIRST
  15.             step = rule.getStep(consumerURL);
  16.             //threshold: 决策阈值（默认-1.0）计算与获取
  17.             threshold = rule.getThreshold(consumerURL);
  18.         } catch (Exception e) {
  19.             logger.error("Failed to get step and threshold info from rule: " + rule, e);
  20.         }
  21.         //刷洗调用器对象 来进行决策服务发现模式
  22.         if (refreshInvoker(step, threshold, rule)) {
  23.             // refresh success, update rule
  24.             setMigrationRule(rule);
  25.         }
  26.     }

22.3.2 服务发现调用器对象的选择（决策服务发现策略）

这里就是关键代码了：通过迁移配置和当前提供者注册信息来决定创建什么类型的调用器对象（Invoker)来为后续服务调用做准备

MigrationRuleHandler的refreshInvoker，注意默认情况下这个step参数为APPLICATION_FIRST

  1. private boolean refreshInvoker(MigrationStep step, Float threshold, MigrationRule newRule) {
  2.        if (step == null || threshold == null) {
  3.            throw new IllegalStateException("Step or threshold of migration rule cannot be null");
  4.        }
  5.        MigrationStep originStep = currentStep;
  7.        if ((currentStep == null || currentStep != step) || !currentThreshold.equals(threshold)) {
  8.            boolean success = true;
  9.            switch (step) {
  10.                case APPLICATION_FIRST:
  11.                    //默认和配置了应用级优先的服务发现则走这里
  12.                    migrationInvoker.migrateToApplicationFirstInvoker(newRule);
  13.                    break;
  14.                case FORCE_APPLICATION:
  15.                //配置了应用级服务发现则走这里
  16.                    success = migrationInvoker.migrateToForceApplicationInvoker(newRule);
  17.                    break;
  18.                case FORCE_INTERFACE:
  19.                //配置了接口级服务发现则走这里
  20.                default:
  21.                    success = migrationInvoker.migrateToForceInterfaceInvoker(newRule);
  22.            }
  24.            if (success) {
  25.                setCurrentStepAndThreshold(step, threshold);
  26.                logger.info("Succeed Migrated to " + step + " mode. Service Name: " + consumerURL.getDisplayServiceKey());
  27.                report(step, originStep, "true");
  28.            } else {
  29.                // migrate failed, do not save new step and rule
  30.                logger.warn("Migrate to " + step + " mode failed. Probably not satisfy the threshold you set "
  31.                        + threshold + ". Please try re-publish configuration if you still after check.");
  32.                report(step, originStep, "false");
  33.            }
  35.            return success;
  36.        }
  37.        // ignore if step is same with previous, will continue override rule for MigrationInvoker
  38.        return true;
  39.    }

可以看到这个代码做了判断的逻辑分别对应了Dubbo3消费者迁移的一个状态逻辑：
三种状态分别如下枚举类型：
当前共存在三种状态，

  • FORCE_INTERFACE（强制接口级）
  • APPLICATION_FIRST（应用级优先）
  • FORCE_APPLICATION（强制应用级）

通过代码我们可以看到默认情况下都会走APPLICATION_FIRST（应用级优先）的策略，这里我们也重点来说 APPLICATION_FIRST（应用级优先）来看下Dubbo3是如何决策使用接口级还是应用级发现模型来兼容迁移的服务的。

22.3.3 应用级优先的服务发现规则逻辑

这个规则就是智能选择应用级还是接口级的代码了，对应类型为MigrationInvoker的migrateToApplicationFirstInvoker方法，接下来我们详细看下：

MigrationInvoker类型的migrateToApplicationFirstInvoker方法： 

  1. @Override
  2.    public void migrateToApplicationFirstInvoker(MigrationRule newRule) {
  3.        CountDownLatch latch = new CountDownLatch(0);
  4.        //刷新接口级服务发现Invoker 
  5.        refreshInterfaceInvoker(latch);
  6.        //刷新应用级服务发现Invoker类型对象
  7.        refreshServiceDiscoveryInvoker(latch);
  9.        // directly calculate preferred invoker, will not wait until address notify
  10.        // calculation will re-occurred when address notify later
  11.        //计算当前使用应用级还是接口级服务发现的Invoker对象
  12.        calcPreferredInvoker(newRule);
  13.    }

22.3.4 刷新接口级服务发现Invoker

MigrationInvoker类型的refreshInterfaceInvoker方法

  1. protected void refreshInterfaceInvoker(CountDownLatch latch) {
  2.         clearListener(invoker);
  3.         if (needRefresh(invoker)) {
  4.             if (logger.isDebugEnabled()) {
  5.                 logger.debug("Re-subscribing interface addresses for interface " + type.getName());
  6.             }
  8.             if (invoker != null) {
  9.                 invoker.destroy();
  10.             }
  11.             invoker = registryProtocol.getInvoker(cluster, registry, type, url);
  12.         }
  13.         setListener(invoker, () -> {
  14.             latch.countDown();
  15.             if (reportService.hasReporter()) {
  16.                 reportService.reportConsumptionStatus(
  17.                     reportService.createConsumptionReport(consumerUrl.getServiceInterface(), consumerUrl.getVersion(), consumerUrl.getGroup(), "interface"));
  18.             }
  19.             if (step == APPLICATION_FIRST) {
  20.                 calcPreferredInvoker(rule);
  21.             }
  22.         });
  23.     }

22.3.5 刷新应用级服务发现Invoker类型对象

MigrationInvoker类型的refreshServiceDiscoveryInvoker方法

  1. protected void refreshServiceDiscoveryInvoker(CountDownLatch latch) {
  2.         clearListener(serviceDiscoveryInvoker);
  3.         if (needRefresh(serviceDiscoveryInvoker)) {
  4.             if (logger.isDebugEnabled()) {
  5.                 logger.debug("Re-subscribing instance addresses, current interface " + type.getName());
  6.             }
  8.             if (serviceDiscoveryInvoker != null) {
  9.                 serviceDiscoveryInvoker.destroy();
  10.             }
  11.             serviceDiscoveryInvoker = registryProtocol.getServiceDiscoveryInvoker(cluster, registry, type, url);
  12.         }
  13.         setListener(serviceDiscoveryInvoker, () -> {
  14.             latch.countDown();
  15.             if (reportService.hasReporter()) {
  16.                 reportService.reportConsumptionStatus(
  17.                     reportService.createConsumptionReport(consumerUrl.getServiceInterface(), consumerUrl.getVersion(), consumerUrl.getGroup(), "app"));
  18.             }
  19.             if (step == APPLICATION_FIRST) {
  20.                 calcPreferredInvoker(rule);
  21.             }
  22.         });
  23.     }

22.3.6 计算当前使用应用级还是接口级服务发现的Invoker对象

MigrationInvoker类型的的calcPreferredInvoker方法

  1. private synchronized void calcPreferredInvoker(MigrationRule migrationRule) {
  2.         if (serviceDiscoveryInvoker == null || invoker == null) {
  3.             return;
  4.         }
  5.         Set<MigrationAddressComparator> detectors = ScopeModelUtil.getApplicationModel(consumerUrl == null ? null : consumerUrl.getScopeModel())
  6.             .getExtensionLoader(MigrationAddressComparator.class).getSupportedExtensionInstances();
  7.         if (CollectionUtils.isNotEmpty(detectors)) {
  8.             // pick preferred invoker
  9.             // the real invoker choice in invocation will be affected by promotion
  10.             if (detectors.stream().allMatch(comparator -> comparator.shouldMigrate(serviceDiscoveryInvoker, invoker, migrationRule))) {
  11.                 this.currentAvailableInvoker = serviceDiscoveryInvoker;
  12.             } else {
  13.                 this.currentAvailableInvoker = invoker;
  14.             }
  15.         }
  16.     }

currentAvailableInvoker是后期服务调用使用的Invoker对象