当前位置：首页 > article >正文

40分钟学 Go 语言高并发：分布式系统理论基础

article 2024/12/5 10:44:17

分布式系统理论基础

一、分布式系统基础概念

理论	关键点	应用场景	重要性
CAP理论	一致性、可用性、分区容忍性	分布式数据库、服务设计	⭐⭐⭐⭐⭐
一致性模型	强一致性、最终一致性、因果一致性	数据同步、状态复制	⭐⭐⭐⭐⭐
分布式事务	2PC、3PC、SAGA	跨服务数据一致性	⭐⭐⭐⭐
故障处理	故障检测、隔离、恢复	系统容错、高可用	⭐⭐⭐⭐⭐

让我们通过代码示例来理解这些概念：

package main

import (
    "context"
    "fmt"
    "log"
    "sync"
    "time"
)

// 模拟分布式节点
type Node struct {
    ID         string
    Data       map[string]string
    mu         sync.RWMutex
    peers      map[string]*Node
    isHealthy  bool
    commitLog  []string
}

// 创建新节点
func NewNode(id string) *Node {
    return &Node{
        ID:        id,
        Data:      make(map[string]string),
        peers:     make(map[string]*Node),
        isHealthy: true,
        commitLog: make([]string, 0),
    }
}

// 添加对等节点
func (n *Node) AddPeer(peer *Node) {
    n.peers[peer.ID] = peer
}

// 2PC第一阶段：准备
func (n *Node) Prepare(key, value string) bool {
    n.mu.Lock()
    defer n.mu.Unlock()

    if !n.isHealthy {
        return false
    }

    // 记录预提交日志
    n.commitLog = append(n.commitLog, fmt.Sprintf("PREPARE:%s:%s", key, value))
    return true
}

// 2PC第二阶段：提交
func (n *Node) Commit(key, value string) bool {
    n.mu.Lock()
    defer n.mu.Unlock()

    if !n.isHealthy {
        return false
    }

    // 执行实际提交
    n.Data[key] = value
    n.commitLog = append(n.commitLog, fmt.Sprintf("COMMIT:%s:%s", key, value))
    return true
}

// 回滚操作
func (n *Node) Rollback(key string) {
    n.mu.Lock()
    defer n.mu.Unlock()

    delete(n.Data, key)
    n.commitLog = append(n.commitLog, fmt.Sprintf("ROLLBACK:%s", key))
}

// 实现两阶段提交协调者
type Coordinator struct {
    nodes []*Node
}

func NewCoordinator(nodes []*Node) *Coordinator {
    return &Coordinator{nodes: nodes}
}

// 执行分布式事务
func (c *Coordinator) ExecuteTransaction(key, value string) bool {
    // 第一阶段：准备
    preparedNodes := make([]*Node, 0)
    for _, node := range c.nodes {
        if node.Prepare(key, value) {
            preparedNodes = append(preparedNodes, node)
        } else {
            // 如果有节点准备失败，回滚已准备的节点
            for _, preparedNode := range preparedNodes {
                preparedNode.Rollback(key)
            }
            return false
        }
    }

    // 第二阶段：提交
    for _, node := range preparedNodes {
        if !node.Commit(key, value) {
            // 实际系统中需要更复杂的恢复机制
            log.Printf("Node %s failed to commit", node.ID)
            return false
        }
    }

    return true
}

// 最终一致性复制
type ReplicationManager struct {
    nodes      []*Node
    retryDelay time.Duration
    maxRetries int
}

func NewReplicationManager(nodes []*Node) *ReplicationManager {
    return &ReplicationManager{
        nodes:      nodes,
        retryDelay: time.Second,
        maxRetries: 3,
    }
}

// 异步复制数据
func (rm *ReplicationManager) ReplicateAsync(ctx context.Context, key, value string) {
    go func() {
        for _, node := range rm.nodes {
            retries := 0
            for retries < rm.maxRetries {
                if node.isHealthy {
                    success := node.Commit(key, value)
                    if success {
                        break
                    }
                }
                retries++
                time.Sleep(rm.retryDelay)
            }
        }
    }()
}

// 故障检测器
type FailureDetector struct {
    nodes         []*Node
    checkInterval time.Duration
    timeout       time.Duration
}

func NewFailureDetector(nodes []*Node) *FailureDetector {
    return &FailureDetector{
        nodes:         nodes,
        checkInterval: time.Second,
        timeout:       time.Second * 3,
    }
}

// 开始故障检测
func (fd *FailureDetector) Start(ctx context.Context) {
    go func() {
        ticker := time.NewTicker(fd.checkInterval)
        defer ticker.Stop()

        for {
            select {
            case <-ticker.C:
                fd.checkNodes()
            case <-ctx.Done():
                return
            }
        }
    }()
}

// 检查节点健康状态
func (fd *FailureDetector) checkNodes() {
    for _, node := range fd.nodes {
        // 在实际系统中，这里会进行网络探测
        if !node.isHealthy {
            log.Printf("Node %s is unhealthy", node.ID)
        }
    }
}

func main() {
    // 创建三个节点
    node1 := NewNode("node1")
    node2 := NewNode("node2")
    node3 := NewNode("node3")

    // 设置对等节点
    node1.AddPeer(node2)
    node1.AddPeer(node3)
    node2.AddPeer(node1)
    node2.AddPeer(node3)
    node3.AddPeer(node1)
    node3.AddPeer(node2)

    nodes := []*Node{node1, node2, node3}

    // 创建协调者
    coordinator := NewCoordinator(nodes)

    // 创建复制管理器
    replicationManager := NewReplicationManager(nodes)

    // 创建故障检测器
    failureDetector := NewFailureDetector(nodes)
    ctx := context.Background()
    failureDetector.Start(ctx)

    // 执行分布式事务
    success := coordinator.ExecuteTransaction("key1", "value1")
    fmt.Printf("Transaction success: %v\n", success)

    // 异步复制数据
    replicationManager.ReplicateAsync(ctx, "key2", "value2")

    // 模拟节点故障
    node2.isHealthy = false
    time.Sleep(time.Second * 2)

    // 检查数据一致性
    for _, node := range nodes {
        fmt.Printf("Node %s data: %v\n", node.ID, node.Data)
    }
}