181｜Dubbo 核心架构：服务暴露/引用、集群容错、路由、负载均衡和 SPI

Consumer 端:
  Proxy(JDK/Javassist) → ClusterInvoker → Directory(Registry) → Router → LoadBalance
  → Filter Chain → Protocol(Invoker) → ExchangeClient → Transport(Netty) → Serialize

Provider 端:
  Transport(Netty) → Deserialize → ExchangeServer → Protocol(Exporter)
  → Filter Chain → Invoker(ServiceBean) → 实际业务实现

// 只加载指定名称的实现
Protocol protocol = ExtensionLoader
    .getExtensionLoader(Protocol.class)
    .getExtension("dubbo");

dubbo=org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol
rest=org.apache.dubbo.rpc.protocol.rest.RestProtocol

public class ProtocolFilterWrapper implements Protocol {
    private final Protocol protocol;
    // Dubbo 自动将真正的 Protocol 实现注入
    public ProtocolFilterWrapper(Protocol protocol) {
        this.protocol = protocol;
    }
}

public class DubboProtocol extends AbstractProtocol {
    // Dubbo 自动注入 ExchangeClient 的 Adaptive 扩展
    private ExchangeClient exchangeClient;
    public void setExchangeClient(ExchangeClient client) {
        this.exchangeClient = client;
    }
}

@SPI("dubbo")
public interface Protocol {
    @Adaptive
    <T> Exporter<T> export(Invoker<T> invoker) throws RpcException;
}

ExtensionLoader.getExtensionLoader(Protocol.class)
  → 扫描 META-INF/dubbo/internal/ 下的配置文件
  → 按 name 映射到对应的 Class
  → 创建实例（包装类自动注入）
  → 注入 setter 依赖（IOC）
  → 返回包装后的实例

ServiceBean.export()
  → doExport() // 组装 URL：从 @DubboService 注解、XML 配置、系统属性等合并参数
  → doExportUrls() // 为每个注册中心和协议生成 URL
  → loadRegistries() // 获取注册中心地址列表
  → doExportUrlsFor1Protocol(protocolConfig)
    → 构建 URL，例如：dubbo://192.168.1.100:20880/com.example.HelloService?version=1.0.0&...
    → 本地暴露：protocol.export(invoker) // 默认 injvm 协议，同 JVM 走本地调用
    → 远程暴露：registryProtocol.export(invoker)
      → 注册中心注册：zkClient.create("/dubbo/com.example.HelloService/providers/...")
      → 协议暴露：DubboProtocol.export(invoker) // 开启 Netty Server 监听 20880

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
    URL url = invoker.getUrl();
    String key = serviceKey(url); // 例如：com.example.HelloService:20880
    // 创建 DubboExporter，持有 invoker 引用
    DubboExporter<T> exporter = new DubboExporter<>(invoker, key, exporterMap);
    exporterMap.put(key, exporter);
    // 开启 Netty Server（如果该端口还没有）
    openServer(url);
    return exporter;
}

dubbo://192.168.1.100:20880/com.example.HelloService
?anyhost=true
&application=demo-provider
&deprecated=false
&dubbo=2.0.2
&dynamic=true
&generic=false
&interface=com.example.HelloService
&methods=sayHello
&pid=12345
&timeout=3000
&version=1.0.0

ReferenceBean.getObject()
  → init()
    → createProxy(map)
      → 判断是本地引用（injvm）还是远程引用
      → 远程引用：protocol.refer(interfaceClass, url)
        → RegistryProtocol.refer()
          → 订阅注册中心：zkClient.subscribe("/dubbo/com.example.HelloService/providers")
          → 获取提供者地址列表 → 构建 Directory
          → Cluster.join(directory) // 将多个 Invoker 聚合成一个 ClusterInvoker
          → 返回代理对象：ProxyFactory.getProxy(clusterInvoker)

public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
    // 1. 获取注册中心实例
    Registry registry = getRegistry(url);
    // 2. 创建 RegistryDirectory，负责服务发现和地址列表维护
    RegistryDirectory<T> directory = new RegistryDirectory<>(type, url);
    directory.setRegistry(registry);
    directory.setProtocol(protocol);
    // 3. 订阅注册中心
    directory.subscribe(subscribeUrl.addParameter(CATEGORY_KEY,
        PROVIDERS_CATEGORY + "," + CONFIGURATORS_CATEGORY + "," + ROUTERS_CATEGORY));
    // 4. 将 Directory 包装为 ClusterInvoker
    return cluster.join(directory);
}

public class proxy0 implements HelloService {
    public String sayHello(String name) {


        Invocation inv = new RpcInvocation("sayHello", new Class[ ]{String.class}, new Object[ ]{name});


        Result result = handler.invoke(inv);
        return result.recreate();
    }
}

ClusterInvoker
  └── Directory（服务目录）
  └── Router（路由器）
  └── LoadBalance（负载均衡）

public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) {
    List<Invoker<T>> copyInvokers = invokers;
    int len = getUrl().getMethodParameter(invocation.getMethodName(), RETRIES_KEY, 2) + 1;
    RpcException le = null;
    List<Invoker<T>> invoked = new ArrayList<>(len);
    for (int i = 0; i < len; i++) {
        // 重试前重新做路由和负载均衡选择
        Invoker<T> invoker = select(loadbalance, invocation, copyInvokers, invoked);
        invoked.add(invoker);
        try {
            Result result = invoker.invoke(invocation);
            return result;
        } catch (RpcException e) {
            le = e;
        }
        // 下一次重试时，从候选列表中移除已失败的
        copyInvokers = invokers;
        if (copyInvokers.size() == invoked.size()) {
            break; // 没有更多可用的提供者了
        }
    }
    throw le; // 全部失败，抛出最后一个异常
}

Invoker 列表
  → MockServiceRouter（Mock 路由，最高优先级）
  → TagRouter（标签路由，用于灰度发布）
  → AppRouter（应用级路由，基于应用名匹配）
  → ServiceRouter（服务级路由，基于服务名匹配）
  → ConditionRouter（条件路由）
  → MeshRouteRouter（Mesh 路由）
  → 筛选后的 Invoker 列表 → 负载均衡

// 示例 1：强制灰度
host = 10.20.153.10 => host = 10.20.153.*

// 示例 2：读写分离
method = find*,list*,get*,query* => host = 172.22.3.91,172.22.3.92
method != find*,list*,get*,query* => host = 172.22.3.93

// 示例 3：黑白名单
host != 10.20.153.10 =>

# 配置示例
tag-router:
  force: false  # force=false 时降级到普通提供者
  tags:
    - name: gray
      addresses: ["192.168.1.101:20880", "192.168.1.102:20880"]

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    int totalWeight = 0;
    boolean sameWeight = true;
    // 1. 计算总权重
    for (int i = 0; i < invokers.size(); i++) {
        int weight = getWeight(invokers.get(i), invocation);
        totalWeight += weight;
        if (sameWeight && i > 0 && weight != getWeight(invokers.get(i-1), invocation)) {
            sameWeight = false;
        }
    }
    // 2. 随机偏移
    if (totalWeight > 0 && !sameWeight) {
        int offset = ThreadLocalRandom.current().nextInt(totalWeight);
        for (Invoker<T> invoker : invokers) {
            offset -= getWeight(invoker, invocation);
            if (offset < 0) {
                return invoker;
            }
        }
    }
    // 3. 权重相同则纯随机
    return invokers.get(ThreadLocalRandom.current().nextInt(invokers.size()));
}

假设 A(weight=5), B(weight=1), C(weight=1)
第1轮: A+=5→5, B+=1→1, C+=1→1 → 选 A, A-=7→-2
第2轮: A+=5→3, B+=1→2, C+=1→2 → 选 A, A-=7→-4
第3轮: A+=5→1, B+=1→3, C+=1→3 → 选 B, B-=7→-4
...

// 配置：对方法 getOrder 的前 2 个参数做哈希
@DubboReference(loadbalance = "consistenthash", parameters = {"hash.arguments", "0,1"})
private OrderService orderService;

ClusterInvoker.invoke()
  → 获取 Directory 中的 Invoker 列表
  → RouterChain.route() 过滤
  → LoadBalance.select() 选中一个
  → 调用 invoke()
  → 如果失败：
    → Failover：重试，重新走 Router + LoadBalance
    → Failfast：直接抛异常
    → Failsafe：吞掉异常

1. helloService.sayHello("world")
   → 代理对象 proxy0.sayHello("world")
   → 构造 RpcInvocation(方法名, 参数类型, 参数值)

2. ClusterInvoker.invoke(invocation)
   → 从 Directory 获取提供者列表
   → Router Chain 路由过滤
   → LoadBalance 选择目标 Invoker

3. 经过 Consumer Filter Chain
   → MonitorFilter 记录调用统计
   → 其他自定义 Filter

4. DubboInvoker.doInvoke()
   → 从连接池获取 ExchangeClient
   → 构造 Request(id, invocation)
   → 发送请求（Netty Channel.writeAndFlush）

5. 等待响应
   → DefaultFuture 阻塞等待（默认超时 1000ms）
   → 收到响应后唤醒

6. NettyServerHandler 收到请求
   → 反序列化 Request
   → 交给 ExchangeHandler 处理

7. 根据 serviceKey 从 exporterMap 中查找 DubboExporter
   → 获取服务端 Invoker

8. 经过 Provider Filter Chain
   → MonitorFilter, ExceptionFilter, 自定义 Filter

9. 反射调用实际业务方法
   → Method.invoke(serviceImpl, args)

10. 构造 Response(result)
    → 序列化响应
    → 通过 Channel 写回消费者

// 提供者
@DubboService(version = "2.0.0")
public class HelloServiceImplV2 implements HelloService {}

// 消费者
@DubboReference(version = "2.0.0")  // 只调用 2.0.0 版本
private HelloService helloService;

# 消费者：只订阅但不注册
dubbo.registry.address=zookeeper://127.0.0.1:2181?register=false

# 提供者：只注册但不订阅
dubbo.registry.address=zookeeper://127.0.0.1:2181?subscribe=false

// 方式 1：CompletableFuture
CompletableFuture<String> future = RpcContext.getServiceContext().asyncCall(
    () -> helloService.sayHello("world")
);
future.whenComplete((result, ex) -> {
    // 处理结果
});

// 方式 2：返回值直接定义为 CompletableFuture
@DubboReference
private HelloService helloService; // 接口方法返回 CompletableFuture<String>

<!-- 为某个服务单独配置线程池 -->
<dubbo:service interface="com.example.SlowService" 
    executors="slowExecutor" />
<dubbo:service interface="com.example.FastService" 
    executors="fastExecutor" />