Nacos 源码阅读

注册中心核心源码

服务注册流程

NacosNamingService.registerInstance(serviceName, ip, port)
  │
  ├─ 组装 Instance 对象
  │
  ├─ NamingHttpProxy / NamingGrpcProxy.reqApi()
  │    │
  │    └─ POST /nacos/v1/ns/instance
  │         │
  │         └─ InstanceController.register()
  │              │
  │              └─ InstanceOperatorClientImpl.registerInstance()
  │                   │
  │                   ├─ 参数校验
  │                   ├─ 设置 instanceId
  │                   ├─ ConsistencyService.put()  ← 数据一致性写入
  │                   └─ 发布 InstanceEvent
  │
  └─ 客户端定时心跳
       └─ BeatReactor.addBeatInfo()
            └─ 每 5s 发送 PUT /nacos/v1/ns/instance/beat

服务发现流程

NacosNamingService.getAllInstances(serviceName)
  │
  ├─ 先查本地缓存 (ServiceInfoHolder)
  │    └─ 有缓存 → 直接返回
  │
  ├─ 无缓存 → 查询集群
  │    └─ GET /nacos/v1/ns/instance/list?serviceName=xxx
  │
  └─ 注册本地监听
       └─ HostReactor.updateServiceNow()
            ├─ 查询全量实例
            └─ 对比增量 → 更新本地缓存

---

AP / CP 切换

核心逻辑

public class ConsistencyServiceFactory {
    public ConsistencyService create(String type) {
        return switch (NacosApplicationConfig.getApMode()) {
            case AP -> new DistroConsistencyServiceImpl();  // Distro 协议
            case CP -> new PersistentConsistencyServiceImpl(); // JRaft 协议
        };
    }
}

特性	AP 模式（Distro）	CP 模式（JRaft）
协议	自研 Distro	Raft 算法
一致性	最终一致性	强一致性
写性能	高（异步复制）	中（Raft 多数派）
选主	无需选举	需 Leader 选举
使用场景	注册中心（默认）	配置中心

Distro 协议核心

public class DistroConsistencyServiceImpl implements ConsistencyService {
    // 本机数据变更 → 同步到其他节点
    public void put(String key, Value value) {
        // 1. 写入本地存储
        distroDataStore.put(key, value);
        // 2. 异步同步到所有远端
        distroSyncTask.addTask(
            new DistroSyncDataTask(key, DistroConfig.getSyncDelayMillis())
        );
        // 3. 发布事件
        notifyListener(key, value);
    }
}

---

健康检查机制

客户端心跳

public class BeatReactor implements Runnable {
    private static final long CLIENT_BEAT_INTERVAL = 5_000L; // 5s

    @Override
    public void run() {
        while (running) {
            for (BeatInfo beatInfo : beatInfos.values()) {
                // 发送心跳
                beatInfo.getClient().sendBeat(beatInfo);
            }
            Thread.sleep(CLIENT_BEAT_INTERVAL);
        }
    }
}

服务端健康检查

public class HealthCheckProcessor {
    // 临时实例：客户端主动心跳
    void processClientBeat(Instance instance) {
        instance.setLastBeat(System.currentTimeMillis());
        // 15s 无心跳 → 标记不健康
        // 30s 无心跳 → 剔除
    }

    // 永久实例：服务端主动探测
    void doHealthCheck(Instance instance) {
        if (instance.isEphemeral()) return; // 临时实例无需服务端探测
        // TCP 端口探测 / HTTP 探测
        boolean healthy = healthChecker.check(instance);
        instance.setHealthy(healthy);
    }
}

---

配置推送

长轮询机制

客户端 → POST /nacos/v1/cs/configs/listener
  │
  ├─ 服务端：检查配置是否有变更
  │    ├─ 无变更 → 挂起请求（最多 30s）
  │    │               ↓
  │    │         配置变更时 → 立即返回变更的 dataId
  │    │
  │    └─ 有变更 → 立即返回
  │
  └─ 客户端收到变更列表 → 逐个拉取最新配置

ConfigService 源码

public class ClientWorker implements Runnable {
    // 长轮询线程
    @Override
    public void run() {
        List<String> changedKeys = checkUpdateDataIds(cacheMap);
        if (!changedKeys.isEmpty()) {
            // 有变更 → 通知监听器
            for (String key : changedKeys) {
                ConfigChangeEvent event = new ConfigChangeEvent(key, getConfig(key));
                configListeners.get(key).forEach(l -> l.receiveConfigInfo(event));
            }
        }
    }
}