第3章：分布式协调

// 传统方式（不推荐）
const (
    DBHost = "mysql.prod.com"  // 硬编码，修改需重新发布
    MaxQPS = 1000
)

// 分布式协调方式（推荐）
type ConfigCenter struct {
    zkConn *zk.Conn
}

func (c *ConfigCenter) GetConfig(key string) (string, error) {
    data, _, err := c.zkConn.Get("/config/" + key)
    return string(data), err
}

func (c *ConfigCenter) WatchConfig(key string, callback func(string)) {
    for {
        data, _, eventCh, err := c.zkConn.GetW("/config/" + key)
        if err != nil {
            continue
        }

        callback(string(data))  // 首次读取

        // 等待变更通知
        event := <-eventCh
        if event.Type == zk.EventNodeDataChanged {
            // 配置变更，递归监听
        }
    }
}

// Leader接收客户端写请求
func (l *Leader) HandleWrite(request WriteRequest) {
    // 1. 生成全局递增的zxid
    zxid := l.GenerateZxid()

    // 2. 发送Proposal给所有Follower
    proposal := Proposal{
        Zxid: zxid,
        Data: request.Data,
    }

    ackCount := 1  // Leader自己先ACK
    for _, follower := range l.followers {
        go func(f *Follower) {
            if f.SendProposal(proposal) {
                ackCount++
            }
        }(follower)
    }

    // 3. 等待过半ACK
    if ackCount > len(l.followers)/2 {
        // 4. 发送Commit给所有Follower
        for _, follower := range l.followers {
            follower.SendCommit(zxid)
        }

        // 5. 返回客户端成功
        return Success
    }

    return Failed
}

package main

import (
    "github.com/samuel/go-zookeeper/zk"
    "time"
)

type ZKClient struct {
    conn *zk.Conn
}

func NewZKClient(servers []string) (*ZKClient, error) {
    conn, _, err := zk.Connect(servers, 10*time.Second)
    if err != nil {
        return nil, err
    }
    return &ZKClient{conn: conn}, nil
}

// 创建持久节点（配置存储）
func (c *ZKClient) CreatePersistent(path, data string) error {
    _, err := c.conn.Create(
        path,
        []byte(data),
        0,  // flags=0 表示持久节点
        zk.WorldACL(zk.PermAll),
    )
    return err
}

// 创建临时节点（服务注册）
func (c *ZKClient) CreateEphemeral(path, data string) error {
    _, err := c.conn.Create(
        path,
        []byte(data),
        zk.FlagEphemeral,  // 临时节点
        zk.WorldACL(zk.PermAll),
    )
    return err
}

// 创建临时顺序节点（分布式锁）
func (c *ZKClient) CreateEphemeralSequential(path, data string) (string, error) {
    return c.conn.Create(
        path,
        []byte(data),
        zk.FlagEphemeral|zk.FlagSequence,  // 临时+顺序
        zk.WorldACL(zk.PermAll),
    )
}

// 配置热更新
type ConfigWatcher struct {
    zkClient *ZKClient
    cache    map[string]string
}

func (w *ConfigWatcher) WatchConfig(key string) {
    path := "/config/" + key

    for {
        // GetW: 获取数据并设置Watch
        data, stat, eventCh, err := w.zkClient.conn.GetW(path)
        if err != nil {
            time.Sleep(1 * time.Second)
            continue
        }

        // 更新缓存
        w.cache[key] = string(data)
        log.Printf("Config updated: %s = %s (version=%d)",
            key, string(data), stat.Version)

        // 等待变更事件
        event := <-eventCh
        log.Printf("Received event: %v", event.Type)

        // Watch是一次性的，需要重新设置（循环继续）
    }
}

// 服务发现
type ServiceDiscovery struct {
    zkClient *ZKClient
    services map[string][]string  // serviceName -> [instances]
}

func (sd *ServiceDiscovery) WatchService(serviceName string) {
    path := "/services/" + serviceName

    for {
        // ChildrenW: 获取子节点并设置Watch
        children, stat, eventCh, err := sd.zkClient.conn.ChildrenW(path)
        if err != nil {
            time.Sleep(1 * time.Second)
            continue
        }

        // 更新服务实例列表
        instances := make([]string, 0)
        for _, child := range children {
            data, _, _ := sd.zkClient.conn.Get(path + "/" + child)
            instances = append(instances, string(data))
        }

        sd.services[serviceName] = instances
        log.Printf("Service instances updated: %v", instances)

        // 等待子节点变更事件
        event := <-eventCh
        if event.Type == zk.EventNodeChildrenChanged {
            log.Printf("Service instances changed")
        }
    }
}

// 客户端连接
conn, sessionEvent, err := zk.Connect(
    []string{"zk1:2181", "zk2:2181", "zk3:2181"},
    10*time.Second,  // sessionTimeout
)

// 监听会话事件
go func() {
    for event := range sessionEvent {
        switch event.State {
        case zk.StateConnecting:
            log.Println("Connecting to ZK...")
        case zk.StateConnected:
            log.Println("Connected to ZK")
        case zk.StateHasSession:
            log.Println("Session established")
        case zk.StateDisconnected:
            log.Println("Disconnected, will retry...")
        case zk.StateExpired:
            log.Println("Session expired, need reconnect")
            // 临时节点会被删除，需要重新注册
        }
    }
}()

type LogEntry struct {
    Term    int64   // Leader的term
    Index   int64   // 日志索引
    Command string  // 实际命令
}

// Leader接收客户端写请求
func (l *Leader) HandleWrite(cmd string) error {
    // 1. 追加到本地日志
    entry := LogEntry{
        Term:    l.currentTerm,
        Index:   l.getLastLogIndex() + 1,
        Command: cmd,
    }
    l.log = append(l.log, entry)

    // 2. 并行复制给所有Follower
    successCount := 1  // Leader自己
    for _, follower := range l.followers {
        go func(f *Follower) {
            if f.AppendEntries(entry) {
                successCount++
            }
        }(follower)
    }

    // 3. 等待多数确认
    if successCount > len(l.followers)/2 {
        // 4. 提交日志
        l.commitIndex = entry.Index

        // 5. 应用到状态机
        l.applyToStateMachine(entry)

        return nil
    }

    return errors.New("replication failed")
}

package main

import (
    "context"
    "go.etcd.io/etcd/client/v3"
    "time"
)

type EtcdClient struct {
    client *clientv3.Client
}

func NewEtcdClient(endpoints []string) (*EtcdClient, error) {
    client, err := clientv3.New(clientv3.Config{
        Endpoints:   endpoints,
        DialTimeout: 5 * time.Second,
    })
    if err != nil {
        return nil, err
    }
    return &EtcdClient{client: client}, nil
}

// 写入数据
func (c *EtcdClient) Put(key, value string) error {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()

    _, err := c.client.Put(ctx, key, value)
    return err
}

// 读取数据
func (c *EtcdClient) Get(key string) (string, error) {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()

    resp, err := c.client.Get(ctx, key)
    if err != nil {
        return "", err
    }

    if len(resp.Kvs) == 0 {
        return "", nil
    }

    return string(resp.Kvs[0].Value), nil
}

// 前缀查询
func (c *EtcdClient) GetPrefix(prefix string) (map[string]string, error) {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()

    resp, err := c.client.Get(ctx, prefix, clientv3.WithPrefix())
    if err != nil {
        return nil, err
    }

    result := make(map[string]string)
    for _, kv := range resp.Kvs {
        result[string(kv.Key)] = string(kv.Value)
    }

    return result, nil
}

// CAS（Compare-And-Swap）操作
func (c *EtcdClient) CAS(key, oldValue, newValue string) (bool, error) {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()

    // 事务：如果value=oldValue，则更新为newValue
    txn := c.client.Txn(ctx)
    txn = txn.If(
        clientv3.Compare(clientv3.Value(key), "=", oldValue),
    ).Then(
        clientv3.OpPut(key, newValue),
    )

    resp, err := txn.Commit()
    if err != nil {
        return false, err
    }

    return resp.Succeeded, nil
}

// 服务注册（使用Lease实现临时节点）
type ServiceRegistry struct {
    client   *clientv3.Client
    leaseID  clientv3.LeaseID
    stopCh   chan struct{}
}

func (sr *ServiceRegistry) Register(serviceName, addr string, ttl int64) error {
    ctx := context.Background()

    // 1. 创建租约
    leaseResp, err := sr.client.Grant(ctx, ttl)
    if err != nil {
        return err
    }
    sr.leaseID = leaseResp.ID

    // 2. 注册服务（绑定租约）
    key := "/services/" + serviceName + "/" + addr
    _, err = sr.client.Put(ctx, key, addr, clientv3.WithLease(sr.leaseID))
    if err != nil {
        return err
    }

    // 3. 自动续约
    sr.stopCh = make(chan struct{})
    go sr.keepAlive()

    return nil
}

func (sr *ServiceRegistry) keepAlive() {
    ctx := context.Background()

    // KeepAlive返回一个channel，定期发送续约响应
    keepAliveCh, err := sr.client.KeepAlive(ctx, sr.leaseID)
    if err != nil {
        log.Printf("KeepAlive error: %v", err)
        return
    }

    for {
        select {
        case resp := <-keepAliveCh:
            if resp == nil {
                log.Println("Lease expired")
                return
            }
            log.Printf("Lease %d keepalive, TTL=%d", resp.ID, resp.TTL)

        case <-sr.stopCh:
            // 主动撤销租约
            sr.client.Revoke(ctx, sr.leaseID)
            return
        }
    }
}

func (sr *ServiceRegistry) Unregister() {
    close(sr.stopCh)
}

// 配置热更新（持续监听）
type ConfigWatcher struct {
    client *clientv3.Client
    cache  map[string]string
}

func (w *ConfigWatcher) WatchConfig(key string) {
    ctx := context.Background()

    // Watch返回一个channel，持续推送事件
    watchCh := w.client.Watch(ctx, key)

    for watchResp := range watchCh {
        for _, event := range watchResp.Events {
            switch event.Type {
            case mvccpb.PUT:
                log.Printf("Config updated: %s = %s",
                    string(event.Kv.Key), string(event.Kv.Value))
                w.cache[string(event.Kv.Key)] = string(event.Kv.Value)

            case mvccpb.DELETE:
                log.Printf("Config deleted: %s", string(event.Kv.Key))
                delete(w.cache, string(event.Kv.Key))
            }
        }
    }
}

// 服务发现（监听前缀）
type ServiceDiscovery struct {
    client   *clientv3.Client
    services map[string][]string
}

func (sd *ServiceDiscovery) WatchService(serviceName string) {
    ctx := context.Background()
    prefix := "/services/" + serviceName + "/"

    // 1. 首次获取所有实例
    resp, _ := sd.client.Get(ctx, prefix, clientv3.WithPrefix())
    instances := make([]string, 0)
    for _, kv := range resp.Kvs {
        instances = append(instances, string(kv.Value))
    }
    sd.services[serviceName] = instances

    // 2. 监听前缀（持续）
    watchCh := sd.client.Watch(ctx, prefix, clientv3.WithPrefix())

    for watchResp := range watchCh {
        for _, event := range watchResp.Events {
            switch event.Type {
            case mvccpb.PUT:
                // 新实例上线
                addr := string(event.Kv.Value)
                sd.services[serviceName] = append(sd.services[serviceName], addr)
                log.Printf("Service instance added: %s", addr)

            case mvccpb.DELETE:
                // 实例下线
                addr := string(event.Kv.Value)
                sd.removeInstance(serviceName, addr)
                log.Printf("Service instance removed: %s", addr)
            }
        }
    }
}

func (sd *ServiceDiscovery) removeInstance(serviceName, addr string) {
    instances := sd.services[serviceName]
    for i, instance := range instances {
        if instance == addr {
            sd.services[serviceName] = append(instances[:i], instances[i+1:]...)
            break
        }
    }
}

// 查询某个revision时的数据（时间旅行）
func (c *EtcdClient) GetAtRevision(key string, revision int64) (string, error) {
    ctx := context.Background()

    resp, err := c.client.Get(ctx, key, clientv3.WithRev(revision))
    if err != nil {
        return "", err
    }

    if len(resp.Kvs) == 0 {
        return "", nil
    }

    return string(resp.Kvs[0].Value), nil
}

// 监听从某个revision开始的变更
func (c *EtcdClient) WatchFromRevision(key string, revision int64) {
    ctx := context.Background()

    // 从指定revision开始监听
    watchCh := c.client.Watch(ctx, key, clientv3.WithRev(revision))

    for watchResp := range watchCh {
        for _, event := range watchResp.Events {
            log.Printf("Event at revision %d: %v",
                event.Kv.ModRevision, event.Type)
        }
    }
}

// 配置中心客户端（etcd实现）
type ConfigCenter struct {
    client *clientv3.Client
    cache  sync.Map  // 本地缓存
}

func NewConfigCenter(endpoints []string) (*ConfigCenter, error) {
    client, err := clientv3.New(clientv3.Config{
        Endpoints:   endpoints,
        DialTimeout: 5 * time.Second,
    })
    if err != nil {
        return nil, err
    }

    cc := &ConfigCenter{client: client}

    // 启动时加载所有配置
    cc.loadAllConfigs()

    // 监听配置变更
    go cc.watchConfigs()

    return cc, nil
}

func (cc *ConfigCenter) loadAllConfigs() error {
    ctx := context.Background()
    resp, err := cc.client.Get(ctx, "/config/", clientv3.WithPrefix())
    if err != nil {
        return err
    }

    for _, kv := range resp.Kvs {
        cc.cache.Store(string(kv.Key), string(kv.Value))
    }

    return nil
}

func (cc *ConfigCenter) watchConfigs() {
    ctx := context.Background()
    watchCh := cc.client.Watch(ctx, "/config/", clientv3.WithPrefix())

    for watchResp := range watchCh {
        for _, event := range watchResp.Events {
            key := string(event.Kv.Key)

            switch event.Type {
            case mvccpb.PUT:
                value := string(event.Kv.Value)
                cc.cache.Store(key, value)
                log.Printf("Config updated: %s = %s", key, value)

                // 触发回调（通知应用层）
                cc.notifyConfigChange(key, value)

            case mvccpb.DELETE:
                cc.cache.Delete(key)
                log.Printf("Config deleted: %s", key)
            }
        }
    }
}

// 获取配置（从本地缓存）
func (cc *ConfigCenter) GetString(key string) string {
    if value, ok := cc.cache.Load("/config/" + key); ok {
        return value.(string)
    }
    return ""
}

func (cc *ConfigCenter) GetInt(key string) int {
    value := cc.GetString(key)
    if value == "" {
        return 0
    }
    intVal, _ := strconv.Atoi(value)
    return intVal
}

// 使用示例
func main() {
    cc, _ := NewConfigCenter([]string{"localhost:2379"})

    // 读取配置
    dbHost := cc.GetString("db.host")
    maxQPS := cc.GetInt("rate_limit.max_qps")

    log.Printf("DB Host: %s, Max QPS: %d", dbHost, maxQPS)

    // 配置变更会自动更新到本地缓存
    select {}
}

// 服务注册
type ServiceRegistry struct {
    client  *clientv3.Client
    leaseID clientv3.LeaseID
    stopCh  chan struct{}
}

func (sr *ServiceRegistry) Register(serviceName, addr string, ttl int64) error {
    ctx := context.Background()

    // 创建租约
    leaseResp, err := sr.client.Grant(ctx, ttl)
    if err != nil {
        return err
    }
    sr.leaseID = leaseResp.ID

    // 注册服务
    key := fmt.Sprintf("/services/%s/%s", serviceName, addr)
    _, err = sr.client.Put(ctx, key, addr, clientv3.WithLease(sr.leaseID))
    if err != nil {
        return err
    }

    // 自动续约
    sr.stopCh = make(chan struct{})
    go sr.keepAlive()

    log.Printf("Service registered: %s -> %s", serviceName, addr)
    return nil
}

func (sr *ServiceRegistry) keepAlive() {
    ctx := context.Background()
    keepAliveCh, _ := sr.client.KeepAlive(ctx, sr.leaseID)

    for {
        select {
        case <-keepAliveCh:
            // 续约成功
        case <-sr.stopCh:
            sr.client.Revoke(ctx, sr.leaseID)
            return
        }
    }
}

func (sr *ServiceRegistry) Unregister() {
    close(sr.stopCh)
}

// 服务发现
type ServiceDiscovery struct {
    client    *clientv3.Client
    services  sync.Map  // serviceName -> []string
    callbacks sync.Map  // serviceName -> []func([]string)
}

func NewServiceDiscovery(endpoints []string) (*ServiceDiscovery, error) {
    client, err := clientv3.New(clientv3.Config{
        Endpoints:   endpoints,
        DialTimeout: 5 * time.Second,
    })
    if err != nil {
        return nil, err
    }

    return &ServiceDiscovery{client: client}, nil
}

func (sd *ServiceDiscovery) Discover(serviceName string) []string {
    prefix := "/services/" + serviceName + "/"

    // 首次获取
    ctx := context.Background()
    resp, _ := sd.client.Get(ctx, prefix, clientv3.WithPrefix())

    instances := make([]string, 0)
    for _, kv := range resp.Kvs {
        instances = append(instances, string(kv.Value))
    }

    sd.services.Store(serviceName, instances)

    // 监听变更
    go sd.watchService(serviceName)

    return instances
}

func (sd *ServiceDiscovery) watchService(serviceName string) {
    prefix := "/services/" + serviceName + "/"
    ctx := context.Background()

    watchCh := sd.client.Watch(ctx, prefix, clientv3.WithPrefix())

    for watchResp := range watchCh {
        instances := sd.getCurrentInstances(serviceName)

        for _, event := range watchResp.Events {
            addr := string(event.Kv.Value)

            switch event.Type {
            case mvccpb.PUT:
                instances = append(instances, addr)
                log.Printf("Service instance added: %s", addr)

            case mvccpb.DELETE:
                instances = sd.removeInstance(instances, addr)
                log.Printf("Service instance removed: %s", addr)
            }
        }

        sd.services.Store(serviceName, instances)
        sd.notifyCallbacks(serviceName, instances)
    }
}

func (sd *ServiceDiscovery) OnChange(serviceName string, callback func([]string)) {
    callbacks, _ := sd.callbacks.LoadOrStore(serviceName, []func([]string){})
    callbacks = append(callbacks.([]func([]string)), callback)
    sd.callbacks.Store(serviceName, callbacks)
}

// 使用示例
func main() {
    // 服务注册
    registry := &ServiceRegistry{client: etcdClient}
    registry.Register("order-service", "192.168.1.10:8080", 10)

    // 服务发现
    discovery, _ := NewServiceDiscovery([]string{"localhost:2379"})
    instances := discovery.Discover("order-service")

    fmt.Printf("Order service instances: %v\n", instances)

    // 监听服务变更
    discovery.OnChange("order-service", func(instances []string) {
        fmt.Printf("Instances updated: %v\n", instances)
    })
}

// Leader选举（etcd实现）
type LeaderElection struct {
    client    *clientv3.Client
    session   *concurrency.Session
    election  *concurrency.Election
    isLeader  bool
    callbacks struct {
        onElected func()
        onLost    func()
    }
}

func NewLeaderElection(endpoints []string, electionKey string) (*LeaderElection, error) {
    client, err := clientv3.New(clientv3.Config{
        Endpoints:   endpoints,
        DialTimeout: 5 * time.Second,
    })
    if err != nil {
        return nil, err
    }

    // 创建Session
    session, err := concurrency.NewSession(client, concurrency.WithTTL(10))
    if err != nil {
        return nil, err
    }

    // 创建Election对象
    election := concurrency.NewElection(session, electionKey)

    return &LeaderElection{
        client:   client,
        session:  session,
        election: election,
    }, nil
}

// 竞选Leader
func (le *LeaderElection) Campaign(ctx context.Context, value string) error {
    // Campaign会阻塞直到成为Leader
    if err := le.election.Campaign(ctx, value); err != nil {
        return err
    }

    le.isLeader = true
    log.Printf("Elected as leader: %s", value)

    // 触发回调
    if le.callbacks.onElected != nil {
        le.callbacks.onElected()
    }

    // 监听Leader变更
    go le.watchLeader(ctx)

    return nil
}

func (le *LeaderElection) watchLeader(ctx context.Context) {
    observeCh := le.election.Observe(ctx)

    for resp := range observeCh {
        if string(resp.Kvs[0].Value) != le.election.Key() {
            // Leader变更
            le.isLeader = false
            log.Println("Lost leadership")

            if le.callbacks.onLost != nil {
                le.callbacks.onLost()
            }
        }
    }
}

// 主动放弃Leader
func (le *LeaderElection) Resign(ctx context.Context) error {
    le.isLeader = false
    return le.election.Resign(ctx)
}

func (le *LeaderElection) IsLeader() bool {
    return le.isLeader
}

func (le *LeaderElection) OnElected(callback func()) {
    le.callbacks.onElected = callback
}

func (le *LeaderElection) OnLost(callback func()) {
    le.callbacks.onLost = callback
}

// 使用示例
func main() {
    nodeID := "node-1"

    election, _ := NewLeaderElection(
        []string{"localhost:2379"},
        "/election/master",
    )

    // 设置回调
    election.OnElected(func() {
        fmt.Println("I am the leader now!")
        // 执行Leader逻辑
    })

    election.OnLost(func() {
        fmt.Println("I lost leadership")
        // 停止Leader逻辑
    })

    // 参与竞选
    ctx := context.Background()
    election.Campaign(ctx, nodeID)
}

// 配置中心服务端
type ConfigServer struct {
    etcdClient *clientv3.Client
}

// 创建/更新配置
func (s *ConfigServer) SetConfig(env, app, key, value string) error {
    ctx := context.Background()

    configKey := fmt.Sprintf("/config/%s/%s/%s", env, app, key)

    // 记录审计日志
    s.auditLog("SET", configKey, value)

    _, err := s.etcdClient.Put(ctx, configKey, value)
    return err
}

// 获取配置
func (s *ConfigServer) GetConfig(env, app, key string) (string, error) {
    ctx := context.Background()

    configKey := fmt.Sprintf("/config/%s/%s/%s", env, app, key)

    resp, err := s.etcdClient.Get(ctx, configKey)
    if err != nil {
        return "", err
    }

    if len(resp.Kvs) == 0 {
        return "", errors.New("config not found")
    }

    return string(resp.Kvs[0].Value), nil
}

// 配置中心客户端
type ConfigClient struct {
    etcdClient *clientv3.Client
    env        string
    app        string
    cache      sync.Map
    listeners  []ConfigListener
}

type ConfigListener interface {
    OnConfigChange(key, value string)
}

func NewConfigClient(endpoints []string, env, app string) (*ConfigClient, error) {
    client, err := clientv3.New(clientv3.Config{
        Endpoints:   endpoints,
        DialTimeout: 5 * time.Second,
    })
    if err != nil {
        return nil, err
    }

    cc := &ConfigClient{
        etcdClient: client,
        env:        env,
        app:        app,
    }

    // 加载所有配置
    cc.loadConfigs()

    // 监听配置变更
    go cc.watchConfigs()

    return cc, nil
}

func (cc *ConfigClient) loadConfigs() error {
    ctx := context.Background()
    prefix := fmt.Sprintf("/config/%s/%s/", cc.env, cc.app)

    resp, err := cc.etcdClient.Get(ctx, prefix, clientv3.WithPrefix())
    if err != nil {
        return err
    }

    for _, kv := range resp.Kvs {
        key := strings.TrimPrefix(string(kv.Key), prefix)
        cc.cache.Store(key, string(kv.Value))
    }

    log.Printf("Loaded %d configs", len(resp.Kvs))
    return nil
}

func (cc *ConfigClient) watchConfigs() {
    ctx := context.Background()
    prefix := fmt.Sprintf("/config/%s/%s/", cc.env, cc.app)

    watchCh := cc.etcdClient.Watch(ctx, prefix, clientv3.WithPrefix())

    for watchResp := range watchCh {
        for _, event := range watchResp.Events {
            key := strings.TrimPrefix(string(event.Kv.Key), prefix)
            value := string(event.Kv.Value)

            switch event.Type {
            case mvccpb.PUT:
                cc.cache.Store(key, value)
                log.Printf("Config updated: %s = %s", key, value)

                // 通知监听器
                for _, listener := range cc.listeners {
                    listener.OnConfigChange(key, value)
                }
            }
        }
    }
}

func (cc *ConfigClient) Get(key string) string {
    if value, ok := cc.cache.Load(key); ok {
        return value.(string)
    }
    return ""
}

func (cc *ConfigClient) GetInt(key string, defaultValue int) int {
    value := cc.Get(key)
    if value == "" {
        return defaultValue
    }
    intVal, _ := strconv.Atoi(value)
    return intVal
}

func (cc *ConfigClient) AddListener(listener ConfigListener) {
    cc.listeners = append(cc.listeners, listener)
}

// 使用示例
type MyApp struct {
    configClient *ConfigClient
    maxQPS       int
}

func (app *MyApp) OnConfigChange(key, value string) {
    if key == "max_qps" {
        newQPS, _ := strconv.Atoi(value)
        app.maxQPS = newQPS
        log.Printf("QPS limit updated to %d", newQPS)
    }
}

func main() {
    // 创建配置客户端
    configClient, _ := NewConfigClient(
        []string{"localhost:2379"},
        "prod",
        "order-service",
    )

    app := &MyApp{
        configClient: configClient,
        maxQPS:       configClient.GetInt("max_qps", 1000),
    }

    // 注册监听器
    configClient.AddListener(app)

    // 读取配置
    dbHost := configClient.Get("db.host")
    dbPort := configClient.GetInt("db.port", 3306)

    log.Printf("DB: %s:%d, Max QPS: %d", dbHost, dbPort, app.maxQPS)

    select {}
}

// ZAB选举规则
func (node *ZKNode) CompareVote(vote1, vote2 Vote) Vote {
    // 1. zxid更大的优先（数据更新）
    if vote1.Zxid > vote2.Zxid {
        return vote1
    }
    if vote2.Zxid > vote1.Zxid {
        return vote2
    }

    // 2. zxid相同，myid更大的优先
    if vote1.MyId > vote2.MyId {
        return vote1
    }
    return vote2
}

// Raft选举规则
func (node *RaftNode) RequestVote(req VoteRequest) bool {
    // 1. term更大的优先
    if req.Term < node.currentTerm {
        return false
    }

    // 2. 已经投过票，拒绝
    if node.votedFor != "" && node.votedFor != req.CandidateId {
        return false
    }

    // 3. 候选者的日志必须至少和自己一样新
    if !node.isLogUpToDate(req.LastLogIndex, req.LastLogTerm) {
        return false
    }

    return true
}

// etcd Watch服务端实现（简化版）
type WatchableStore struct {
    mu       sync.RWMutex
    watchers map[string][]*WatchStream  // key -> streams
    revisions map[int64]*Event          // revision -> event
}

func (ws *WatchableStore) Watch(key string, startRevision int64) *WatchStream {
    ws.mu.Lock()
    defer ws.mu.Unlock()

    stream := &WatchStream{
        key:    key,
        eventC: make(chan *Event, 100),
    }

    ws.watchers[key] = append(ws.watchers[key], stream)

    // 发送历史事件（从startRevision开始）
    go func() {
        for rev := startRevision; ; rev++ {
            if event, ok := ws.revisions[rev]; ok {
                if event.Key == key {
                    stream.eventC <- event
                }
            } else {
                break
            }
        }
    }()

    return stream
}

func (ws *WatchableStore) Notify(key string, event *Event) {
    ws.mu.RLock()
    defer ws.mu.RUnlock()

    // 保存事件（MVCC）
    ws.revisions[event.Revision] = event

    // 通知所有Watch该key的stream
    for _, stream := range ws.watchers[key] {
        select {
        case stream.eventC <- event:
        default:
            // channel满了，跳过（客户端消费慢）
        }
    }
}

// ZK Watch（一次性）
for {
    data, _, eventCh, _ := zkConn.GetW("/config/key")
    handleData(data)

    event := <-eventCh
    // 需要重新设置Watch（循环继续）
}

// etcd Watch（持续）
watchCh := etcdClient.Watch(ctx, "/config/key")
for watchResp := range watchCh {
    for _, event := range watchResp.Events {
        handleEvent(event)
    }
    // 无需重新设置，持续接收
}

type ResilientClient struct {
    endpoints []string
    client    *clientv3.Client
    mu        sync.RWMutex
}

func (rc *ResilientClient) ensureConnected() error {
    rc.mu.Lock()
    defer rc.mu.Unlock()

    // 检查连接状态
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()

    _, err := rc.client.Get(ctx, "health_check")
    if err == nil {
        return nil  // 连接正常
    }

    // 重新连接
    log.Println("Reconnecting to etcd...")
    newClient, err := clientv3.New(clientv3.Config{
        Endpoints:   rc.endpoints,
        DialTimeout: 5 * time.Second,
    })
    if err != nil {
        return err
    }

    rc.client.Close()
    rc.client = newClient
    log.Println("Reconnected successfully")

    return nil
}

func (rc *ResilientClient) Get(key string) (string, error) {
    // 自动重连
    if err := rc.ensureConnected(); err != nil {
        return "", err
    }

    ctx := context.Background()
    resp, err := rc.client.Get(ctx, key)
    if err != nil {
        return "", err
    }

    if len(resp.Kvs) == 0 {
        return "", nil
    }

    return string(resp.Kvs[0].Value), nil
}

// Session自动管理
conn, sessionEvent, _ := zk.Connect(
    []string{"localhost:2181"},
    10*time.Second,  // sessionTimeout
)

// 自动心跳，无需手动处理
// 连接断开 → Session过期 → 临时节点删除

// 监听Session事件
go func() {
    for event := range sessionEvent {
        switch event.State {
        case zk.StateConnected:
            log.Println("Connected")
        case zk.StateExpired:
            log.Println("Session expired")
            // 需要重新连接，重新注册临时节点
        }
    }
}()