Dubbo 与 SOFA RPC 的 ZooKeeper 断连容灾机制深度对比

// Dubbo 2.7.x: AbstractRegistry.java
public abstract class AbstractRegistry implements Registry {

    // ====== 内存缓存 (第1层) ======
    // Properties 是线程安全的，key=service, value=provider列表
    private final Properties properties = new Properties();

    // 文件缓存路径
    private File file;

    public AbstractRegistry(URL url) {
        // 启动时从文件恢复
        this.file = new File(System.getProperty("user.home")
            + "/.dubbo/dubbo-registry-" + url.getParameter("application")
            + "-" + url.getAddress().replace(':', '-') + ".cache");

        if (file.exists()) {
            // 从文件加载到内存 Properties
            properties.load(new FileInputStream(file));
        }
    }

    // ====== 文件缓存 (第2层) ======
    // 异步保存到文件
    private void saveProperties(URL url) {
        String key = url.getServiceKey();  // com.xxx.OrderService
        String value = getCacheUrls(url);  // dubbo://10.0.0.1:20880/...,...

        properties.setProperty(key, value);

        // 异步写文件，不阻塞主流程
        long version = url.getParameter("timestamp", 0L);
        if (version > lastCacheChanged.get()) {
            // 启动一个单线程的 ScheduledExecutor 异步保存
            newSaveThread.schedule(() -> {
                synchronized (file) {
                    properties.store(new FileOutputStream(file), "Dubbo Registry Cache");
                }
            }, 500, TimeUnit.MILLISECONDS);  // 500ms 消抖
        }
    }

    // ☆ 最关键的方法: 获取缓存的 Provider 列表 ☆
    // 这个方法不访问 ZK，只读内存!
    public List<URL> getCacheUrls(URL url) {
        for (Map.Entry<Object, Object> entry : properties.entrySet()) {
            String key = (String) entry.getKey();
            if (key.equals(url.getServiceKey())) {
                // 从文件中解析 Provider URL 列表
                return URL.parseURL((String) entry.getValue(), null);
            }
        }
        return Collections.emptyList();
    }
}

// FailbackRegistry.java
public abstract class FailbackRegistry extends AbstractRegistry {

    // ====== 三个失败重试集合 ======
    private final Set<URL> failedRegistered = new ConcurrentHashSet<>();     // 注册失败
    private final Set<URL> failedUnregistered = new ConcurrentHashSet<>();   // 注销失败
    private final ConcurrentMap<URL, NotifyListener> failedSubscribed = ...; // 订阅失败

    // 重试定时器: 每隔 retryPeriod (默认 5000ms) 重试
    private final ScheduledExecutorService retryExecutor =
        Executors.newScheduledThreadPool(1);

    public FailbackRegistry(URL url) {
        super(url);
        // 启动定时重试
        this.retryPeriod = url.getParameter("retry.period", 5000);
        retryExecutor.scheduleWithFixedDelay(() -> {
            try {
                retry();  // 重试所有失败的操作
            } catch (Throwable t) {
                log.error("Unexpected error in retry", t);
            }
        }, retryPeriod, retryPeriod, TimeUnit.MILLISECONDS);
    }

    // ====== 每次写操作都套了 try-catch-failback ======
    @Override
    public void register(URL url) {
        try {
            doRegister(url);  // → ZookeeperRegistry 实际写 ZK
            failedRegistered.remove(url);
        } catch (Exception e) {
            // ☆ 失败不抛异常，加入重试队列 ☆
            failedRegistered.add(url);
            log.warn("Failed to register {}, will retry", url, e);
        }
    }

    @Override
    public void subscribe(URL url, NotifyListener listener) {
        try {
            doSubscribe(url, listener);  // → ZookeeperRegistry 实际注册 Watcher
            failedSubscribed.remove(url);
        } catch (Exception e) {
            failedSubscribed.put(url, listener);  // 加入重试队列
            log.warn("Failed to subscribe {}, will retry", url, e);
        }
    }

    // ====== 重试逻辑 ======
    @Override
    protected void retry() {
        // 重试失败的注册
        Iterator<URL> it = failedRegistered.iterator();
        while (it.hasNext()) {
            URL url = it.next();
            try {
                doRegister(url);
                it.remove();
            } catch (Exception e) {
                log.warn("Retry register still failed: {}", url);
            }
        }

        // 重试失败的订阅 (最重要!)
        // Watcher 丢失后需要重新注册
        for (Map.Entry<URL, NotifyListener> entry : failedSubscribed.entrySet()) {
            try {
                doSubscribe(entry.getKey(), entry.getValue());
                failedSubscribed.remove(entry.getKey());
            } catch (Exception e) {
                log.warn("Retry subscribe still failed");
            }
        }
    }
}

// ZookeeperRegistry.java (Dubbo 2.7.x)
public class ZookeeperRegistry extends FailbackRegistry {

    private CuratorFramework zkClient;  // Apache Curator 封装

    @Override
    protected void doSubscribe(URL url, NotifyListener listener) {
        String path = toServicePath(url);  // /dubbo/com.xxx.OrderService/providers

        // 创建一个 Cache 对象 (Curator 的 TreeCache / PathChildrenCache)
        // Cache 内部处理了 Watcher 的自动重新注册!
        PathChildrenCache cache = pathChildrenCacheMap.computeIfAbsent(path, p -> {
            PathChildrenCache c = new PathChildrenCache(zkClient, p, true);
            c.getListenable().addListener((client, event) -> {
                // 子节点变化回调
                List<URL> urls = toUrlsWithEmpty(client, path, event.getData());
                // 更新内部 Provider 列表 → 通知 Dubbo 的 NotifyListener
                notify(url, listener, urls);
                // 同时更新文件缓存
                saveProperties(url);
            });
            return c;
        });
        cache.start();
    }

    // ====== Curator 连接状态监听 ======
    private void addStateListener() {
        zkClient.getConnectionStateListenable().addListener((client, state) -> {
            switch (state) {
                case RECONNECTED:
                    // ☆ 重连成功: 核心恢复逻辑 ☆
                    log.info("ZK reconnected, recovering...");
                    recover();
                    break;

                case LOST:
                    // Session 过期
                    // Curator 内部会自动重建连接 (新的 sessionId)
                    // 但所有 Watcher 都丢失了 → 需要全量重建
                    log.warn("ZK session expired!");
                    // recover() 会在 RECONNECTED 中触发
                    break;

                case SUSPENDED:
                    // 断连中 → 不 panic，等待 Curator 自动重连
                    // 内存缓存中的 Provider 列表完全可用
                    log.info("ZK connection suspended, waiting for reconnect...");
                    break;
            }
        });
    }

    // ☆ 恢复方法: 重连后必须做的三件事 ☆
    @Override
    protected void recover() {
        // 1. 重新注册本机的 Provider URL
        Set<URL> recoverRegistered = new HashSet<>(getRegistered());
        if (!recoverRegistered.isEmpty()) {
            log.info("Recovering registered URLs: {}", recoverRegistered.size());
            for (URL url : recoverRegistered) {
                try {
                    // 先检查 ZK 上是否还存在 (可能已被自动清理)
                    if (zkClient.checkExists().forPath(toUrlPath(url)) == null) {
                        doRegister(url);  // 重新创建临时节点
                    }
                } catch (Exception e) {
                    failedRegistered.add(url);
                }
            }
        }

        // 2. 重新订阅: 重新拉取全量 Provider 列表 + 重新注册 Watcher
        Set<URL> recoverSubscribed = new HashSet<>(getSubscribed().keySet());
        if (!recoverSubscribed.isEmpty()) {
            log.info("Recovering subscribed URLs: {}", recoverSubscribed.size());
            for (URL url : recoverSubscribed) {
                try {
                    // 重新订阅 (内部会全量拉取 + 重新设置 Watcher)
                    doSubscribe(url, getSubscribed().get(url));
                } catch (Exception e) {
                    failedSubscribed.put(url, getSubscribed().get(url));
                }
            }
        }
    }
}

// SOFA RPC: ZkClient.java (简化版)
public class ZkClient {
    private ZooKeeper zk;
    private volatile State state = State.DISCONNECTED;
    private final List<StateListener> stateListeners = new CopyOnWriteArrayList<>();

    // 连接状态枚举
    enum State {
        DISCONNECTED,  // 未连接
        CONNECTING,    // 连接中
        CONNECTED,     // 已连接
        RECONNECTING,  // 重连中
        CLOSED         // 已关闭
    }

    public ZkClient(String connectString, int sessionTimeout) {
        connect(); // 首次连接
    }

    private void connect() {
        this.zk = new ZooKeeper(connectString, sessionTimeout, event -> {
            switch (event.getState()) {
                case SyncConnected:
                    state = State.CONNECTED;
                    // 通知所有 StateListener: 连接成功
                    for (StateListener listener : stateListeners) {
                        listener.onConnected();
                    }
                    break;

                case Disconnected:
                    state = State.RECONNECTING;
                    // ☆ 注意: SOFA 在断连时通知，但不 panic
                    //         因为 AddressHolder 仍持有有效数据
                    for (StateListener listener : stateListeners) {
                        listener.onDisconnected();
                    }
                    break;

                case Expired:
                    state = State.DISCONNECTED;
                    // Session 过期: 通知上层需要重建
                    for (StateListener listener : stateListeners) {
                        listener.onSessionExpired();
                    }
                    // 自动重连
                    reconnect();
                    break;
            }
        });
    }

    // ☆ 断连自动重连 ☆
    private void reconnect() {
        state = State.CONNECTING;
        try {
            // 关闭旧连接
            zk.close();
        } catch (Exception ignored) {}

        while (state != State.CLOSED) {
            try {
                // 重新建立 ZooKeeper 连接
                connect();
                break;
            } catch (Exception e) {
                log.warn("ZK reconnect failed, retrying...");
                Thread.sleep(3000);  // 固定 3s 重试
            }
        }
    }
}

// SOFA RPC: ZookeeperRegistry.java
public class ZookeeperRegistry extends Registry {

    private ZkClient zkClient;
    private final Map<String, List<ProviderInfoListener>> providerListeners = ...;

    @Override
    public void subscribe(ConsumerConfig config) {
        String interfaceId = config.getInterfaceId();
        String providerPath = "/sofa-rpc/" + interfaceId + "/providers";

        // 1. 全量拉取 Provider 列表
        List<String> children = zkClient.getChildren(providerPath);
        List<ProviderInfo> providers = convertToProviderInfos(children);

        // 2. 交给 AddressHolder 持有
        AddressHolder holder = getAddressHolder(interfaceId);
        holder.updateAllProviders(providers);

        // 3. 注册 Watcher: 监听子节点变化 (Provider 上线/下线)
        zkClient.watchChildChanges(providerPath, (parentPath, currentChildren) -> {
            List<ProviderInfo> newProviders = convertToProviderInfos(currentChildren);
            Set<ProviderInfo> added = diff(newProviders, holder.getAllProviders());
            Set<ProviderInfo> removed = diff(holder.getAllProviders(), newProviders);

            // 增量更新 AddressHolder
            holder.addProviders(added);
            holder.removeProviders(removed);

            // 通知所有 ProviderInfoListener
            notifyProviderListeners(interfaceId, added, removed);
        });

        // 4. 注册连接状态监听
        zkClient.addStateListener(new StateListener() {
            @Override
            public void onConnected() {
                log.info("ZK connected");
            }

            @Override
            public void onDisconnected() {
                // ☆ 关键: 断连时什么都不做!
                //   AddressHolder 继续持有当前列表
                //   消费者感知不到
                log.info("ZK disconnected, holding cached providers: {}",
                    holder.getProviderCount());
            }

            @Override
            public void onSessionExpired() {
                // Session 过期: 所有临时节点已丢失
                log.warn("ZK session expired!");
            }
        });
    }

    // ☆ 重连后的恢复: 与 Curator 不同，SOFA 需要手动处理 ☆
    private void recoverAfterReconnect() {
        // 遍历所有已订阅的服务
        for (String interfaceId : subscribedServices) {
            try {
                // 1. 全量拉取最新 Provider 列表
                String path = "/sofa-rpc/" + interfaceId + "/providers";
                List<String> children = zkClient.getChildren(path);
                List<ProviderInfo> providers = convertToProviderInfos(children);

                // 2. 更新 AddressHolder
                AddressHolder holder = getAddressHolder(interfaceId);
                holder.updateAllProviders(providers);

                // 3. 重新注册 Watcher (一次性机制!)
                zkClient.watchChildChanges(path, ...);

                log.info("Recovered subscription for {}", interfaceId);
            } catch (Exception e) {
                log.error("Recover failed for {}", interfaceId, e);
            }
        }
    }
}

// SOFA RPC 特有的: 对每个 Provider 节点单独注册 Watcher
private void watchProviderNode(String providerPath, ProviderInfo info) {
    // 监听单个 Provider 节点的数据变化
    zkClient.watchData(providerPath, (path, data) -> {
        // Provider 的权重、warmup 等属性变化了
        ProviderInfo updated = parseProviderInfo(data);
        AddressHolder holder = getAddressHolder(updated.getInterfaceId());
        holder.updateProvider(updated);  // 增量更新这一个 Provider
        // ↓
        // 负载均衡组件拿到新的权重，重新计算权重列表
    });
}

// ===== Dubbo: 框架级内置重试 =====
// FailbackRegistry 对所有 ZK 操作都有 try-catch-failback
register(url)     → 失败 → failedRegistered.add(url)     → 定时重试
subscribe(url)    → 失败 → failedSubscribed.put(url)      → 定时重试
unregister(url)   → 失败 → failedUnregistered.add(url)    → 定时重试

// 结论: 开发者完全不用关心 ZK 临时故障

// ===== SOFA RPC: 抛给上层处理 =====
// ZookeeperRegistry 的操作如果失败会抛出异常
subscribe(config)  → ZK 挂了 → 抛异常 → 上层业务代码决定怎么处理
register(provider) → ZK 挂了 → 抛异常 → SOFA 的 ProviderBootstrap 处理

// SOFA 的 ProviderBootstrap 中有内置重试:
public class ProviderBootstrap {
    private void doRegister() {
        // 首次注册
        try {
            registry.register(providerConfig);
        } catch (Exception e) {
            // 失败后定时重试
            scheduledExecutor.scheduleWithFixedDelay(() -> {
                try {
                    registry.register(providerConfig);
                } catch (Exception ex) {
                    log.warn("Provider register retry failed");
                }
            }, 5, 5, TimeUnit.SECONDS);
        }
    }
}