第1章：分布式一致性

// 强一致性示例
func Transfer(from, to string, amount int) error {
    // 所有节点必须看到相同的结果

    tx := db.Begin()

    // 1. 扣减from账户
    tx.Exec("UPDATE accounts SET balance = balance - ? WHERE id = ?", amount, from)

    // 2. 增加to账户
    tx.Exec("UPDATE accounts SET balance = balance + ? WHERE id = ?", amount, to)

    // 3. 提交事务（原子性）
    return tx.Commit()
}

// 读操作总能读到最新的写入
balance := GetBalance("account_A")  // 总是返回最新值

// 高可用示例
func GetBalance(accountID string) (int, error) {
    // 从任意节点读取（可能不是最新值）

    // 优先本地缓存
    if balance, ok := localCache.Get(accountID); ok {
        return balance, nil
    }

    // 从最近的节点读取
    for _, node := range nearbyNodes {
        if balance, err := node.Get(accountID); err == nil {
            return balance, nil  // 立即返回，不保证最新
        }
    }

    return 0, errors.New("all nodes failed")
}

// ZooKeeper写入（CP系统）
func WriteToZK(path string, data []byte) error {
    conn, _, err := zk.Connect([]string{"localhost:2181"}, time.Second)
    if err != nil {
        return err
    }

    // 必须写入多数节点才返回成功
    _, err = conn.Create(path, data, 0, zk.WorldACL(zk.PermAll))
    if err != nil {
        // 网络分区时可能失败（牺牲可用性）
        return err  //  拒绝服务
    }

    return nil  //  保证强一致
}

// Cassandra写入（AP系统）
func WriteToCassandra(key string, value string) error {
    session := cluster.CreateSession()

    // 只需写入部分节点即可返回（牺牲一致性）
    err := session.Query(
        "INSERT INTO data (key, value) VALUES (?, ?)",
        key, value,
    ).Consistency(gocql.One).Exec()  // ONE：只需1个节点成功

    if err != nil {
        return err
    }

    return nil  //  高可用，但可能有节点数据不一致
}

// 读操作
func ReadFromCassandra(key string) (string, error) {
    var value string

    // 从任意节点读取（可能是旧数据）
    err := session.Query(
        "SELECT value FROM data WHERE key = ?",
        key,
    ).Consistency(gocql.One).Scan(&value)

    return value, err  // 可能读到旧值（最终一致性）
}

// 核心数据用CP，非核心数据用AP
type OrderSystem struct {
    orderDB    *MySQL      // CP：订单数据（强一致）
    cacheDB    *Redis      // AP：缓存（最终一致）
    searchDB   *ES         // AP：搜索（最终一致）
}

func (s *OrderSystem) CreateOrder(order *Order) error {
    // 1. 写入MySQL（CP，保证一致性）
    if err := s.orderDB.Insert(order); err != nil {
        return err
    }

    // 2. 异步更新缓存（AP，允许短暂不一致）
    go func() {
        s.cacheDB.Set(order.ID, order)
        s.searchDB.Index(order)
    }()

    return nil
}

// 基本可用：降级策略
func GetUserInfo(userID string) (*User, error) {
    // 尝试从主库读取
    user, err := masterDB.GetUser(userID)
    if err == nil {
        return user, nil
    }

    // 主库失败，尝试从库（可能数据延迟）
    user, err = slaveDB.GetUser(userID)
    if err == nil {
        return user, nil  // 基本可用：返回可能旧的数据
    }

    // 从库也失败，返回缓存
    user, err = cache.GetUser(userID)
    if err == nil {
        return user, nil  // 基本可用：返回缓存数据
    }

    // 完全失败
    return nil, errors.New("service unavailable")
}

// 订单状态机（软状态）
type OrderStatus string

const (
    StatusPending    OrderStatus = "pending"     // 软状态
    StatusPaying     OrderStatus = "paying"      // 软状态
    StatusPaid       OrderStatus = "paid"        // 最终状态
    StatusCancelled  OrderStatus = "cancelled"   // 最终状态
)

type Order struct {
    ID        string
    Status    OrderStatus
    CreatedAt time.Time
}

// 订单状态转换（允许中间状态）
func (o *Order) Process() {
    switch o.Status {
    case StatusPending:
        // 软状态：等待支付
        if time.Since(o.CreatedAt) > 30*time.Minute {
            o.Status = StatusCancelled  // 超时取消
        }

    case StatusPaying:
        // 软状态：支付处理中
        // 异步查询支付结果
        go func() {
            result := queryPaymentResult(o.ID)
            if result.Success {
                o.Status = StatusPaid  // 达到最终状态
            } else {
                o.Status = StatusCancelled
            }
        }()
    }
}

type Data struct {
    Value     string
    Version   int64      // 版本号
    Timestamp time.Time  // 时间戳
}

// 冲突解决：Last Write Wins
func Merge(data1, data2 Data) Data {
    if data1.Timestamp.After(data2.Timestamp) {
        return data1  // 后写入的获胜
    }
    return data2
}

type VectorClock map[string]int64

// 示例：
// Node1: {Node1: 5, Node2: 3, Node3: 2}
// Node2: {Node1: 4, Node2: 6, Node3: 2}

func (vc VectorClock) HappensBefore(other VectorClock) bool {
    // 判断因果关系
    for node, timestamp := range vc {
        if timestamp > other[node] {
            return false
        }
    }
    return true
}

// 节点间周期性同步数据
func (n *Node) GossipSync() {
    ticker := time.NewTicker(1 * time.Second)

    for range ticker.C {
        // 随机选择一个节点同步
        peer := n.SelectRandomPeer()

        // 交换数据
        myData := n.GetAllData()
        peerData := peer.GetAllData()

        // 合并数据（解决冲突）
        mergedData := Merge(myData, peerData)

        // 更新本地数据
        n.UpdateData(mergedData)
    }
}

// 使用BASE理论的库存系统
type InventorySystem struct {
    db    *MySQL      // 持久化存储
    cache *Redis      // 缓存
    mq    *Kafka      // 消息队列
}

// 1. 基本可用：优先扣减缓存
func (s *InventorySystem) DeductStock(productID string, quantity int) error {
    // 1. 从缓存扣减（快速响应）
    remaining, err := s.cache.DecrBy(productID, quantity)
    if err != nil {
        return err
    }

    if remaining < 0 {
        // 库存不足，回滚
        s.cache.IncrBy(productID, quantity)
        return errors.New("insufficient stock")
    }

    // 2. 发送消息到MQ（异步更新DB）
    event := StockDeductEvent{
        ProductID: productID,
        Quantity:  quantity,
        Timestamp: time.Now(),
    }
    s.mq.Publish("stock.deduct", event)

    return nil  // 基本可用：缓存扣减成功即返回
}

// 3. 最终一致性：异步消费MQ，更新DB
func (s *InventorySystem) ConsumeStockEvents() {
    s.mq.Subscribe("stock.deduct", func(event StockDeductEvent) {
        // 更新数据库
        err := s.db.Exec(
            "UPDATE inventory SET stock = stock - ? WHERE product_id = ?",
            event.Quantity, event.ProductID,
        )

        if err != nil {
            // 重试机制
            s.mq.Retry(event)
        }
    })
}

// 4. 定时对账（保证最终一致）
func (s *InventorySystem) Reconcile() {
    ticker := time.NewTicker(1 * time.Hour)

    for range ticker.C {
        // 对比缓存和数据库
        products := s.db.GetAllProducts()

        for _, product := range products {
            cacheStock := s.cache.Get(product.ID)
            dbStock := product.Stock

            if cacheStock != dbStock {
                // 发现不一致，以数据库为准
                s.cache.Set(product.ID, dbStock)
                log.Printf("Reconciled: %s, cache=%d, db=%d",
                    product.ID, cacheStock, dbStock)
            }
        }
    }
}

type RaftNode struct {
    id       string
    state    NodeState  // Leader/Follower/Candidate
    term     int64      // 任期号
    log      []LogEntry
    commitIndex int
}

type NodeState int

const (
    Follower NodeState = iota
    Candidate
    Leader
)

// 选举超时，发起选举
func (n *RaftNode) StartElection() {
    n.state = Candidate
    n.term++
    votes := 1  // 投自己一票

    // 向其他节点请求投票
    for _, peer := range n.peers {
        go func(p *RaftNode) {
            if p.RequestVote(n.term, n.id) {
                votes++
            }
        }(peer)
    }

    // 获得多数票
    if votes > len(n.peers)/2 {
        n.state = Leader
        n.StartHeartbeat()
    }
}

// Leader定期发送心跳
func (n *RaftNode) StartHeartbeat() {
    ticker := time.NewTicker(100 * time.Millisecond)

    for n.state == Leader {
        <-ticker.C
        for _, peer := range n.peers {
            go peer.AppendEntries(n.term, n.log)
        }
    }
}

package main

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

// 配置中心客户端
type ConfigCenter struct {
    client *clientv3.Client
}

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

    if err != nil {
        return nil, err
    }

    return &ConfigCenter{client: client}, nil
}

// 读取配置（强一致性）
func (c *ConfigCenter) GetConfig(key string) (string, error) {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()

    // 从Leader读取，保证强一致性
    resp, err := c.client.Get(ctx, key, clientv3.WithSerializable())
    if err != nil {
        return "", err
    }

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

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

// 更新配置
func (c *ConfigCenter) SetConfig(key, value string) error {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()

    // 写入etcd（需要多数节点确认）
    _, err := c.client.Put(ctx, key, value)
    return err
}

// 监听配置变化
func (c *ConfigCenter) WatchConfig(key string, callback func(string)) {
    watchChan := c.client.Watch(context.Background(), key)

    for resp := range watchChan {
        for _, event := range resp.Events {
            newValue := string(event.Kv.Value)
            callback(newValue)  // 通知配置变更
        }
    }
}

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

    // 设置配置
    cc.SetConfig("/app/db/host", "mysql.example.com")

    // 读取配置
    host, _ := cc.GetConfig("/app/db/host")
    log.Println("DB Host:", host)

    // 监听配置变化
    go cc.WatchConfig("/app/db/host", func(newValue string) {
        log.Println("Config changed:", newValue)
        // 重新加载配置
    })
}

package main

import (
    "github.com/go-redis/redis/v8"
    "context"
    "time"
)

// 分布式缓存
type DistributedCache struct {
    cluster *redis.ClusterClient
}

func NewDistributedCache(addrs []string) *DistributedCache {
    cluster := redis.NewClusterClient(&redis.ClusterOptions{
        Addrs: addrs,

        // AP配置：优先可用性
        ReadOnly:       true,   // 允许从slave读取
        RouteByLatency: true,   // 路由到延迟最低的节点
    })

    return &DistributedCache{cluster: cluster}
}

// 写入缓存
func (c *DistributedCache) Set(key, value string, ttl time.Duration) error {
    ctx := context.Background()

    // 写入master（异步复制到slave）
    return c.cluster.Set(ctx, key, value, ttl).Err()
}

// 读取缓存（可能读到旧值）
func (c *DistributedCache) Get(key string) (string, error) {
    ctx := context.Background()

    // 从最近的节点读取（可能是slave，数据可能延迟）
    value, err := c.cluster.Get(ctx, key).Result()
    if err == redis.Nil {
        return "", nil  // 键不存在
    }

    return value, err
}

// 使用示例
func main() {
    cache := NewDistributedCache([]string{
        "redis-1:6379",
        "redis-2:6379",
        "redis-3:6379",
    })

    // 写入
    cache.Set("user:1001", "Alice", 1*time.Hour)

    // 立即读取（可能从master读）
    value1, _ := cache.Get("user:1001")
    println("Read1:", value1)  // "Alice"

    // 从另一个地域读取（可能从slave读，可能延迟）
    time.Sleep(10 * time.Millisecond)
    value2, _ := cache.Get("user:1001")
    println("Read2:", value2)  // 可能是空（复制延迟）

    // 等待复制完成
    time.Sleep(100 * time.Millisecond)
    value3, _ := cache.Get("user:1001")
    println("Read3:", value3)  // "Alice"（最终一致）
}

// Cassandra写入（Quorum）
consistency := gocql.Quorum  // W=2（写2个副本）

err := session.Query(
    "INSERT INTO users (id, name) VALUES (?, ?)",
    userID, name,
).Consistency(consistency).Exec()

// 只要多数节点确认，数据不会丢失

// 定期Merkle Tree比对
func (n *Node) AntiEntropy() {
    for _, peer := range n.peers {
        myTree := n.BuildMerkleTree()
        peerTree := peer.BuildMerkleTree()

        diff := myTree.Diff(peerTree)

        // 同步差异数据
        for _, key := range diff {
            n.SyncKey(peer, key)
        }
    }
}