面试题：Go 并发设计模式在分布式系统中的应用优化

设计思路

任务调度模式选择：选择流水线模式来处理任务，将任务处理过程拆分成多个阶段，每个阶段由不同的 goroutine 负责，提高并行处理能力。
节点间通信：使用消息队列（如 NATS、RabbitMQ 等）作为节点间通信的桥梁。每个节点从消息队列中获取任务，处理完成后将结果发送回消息队列或者发送给下一个处理阶段的节点。
负载均衡：
- 基于消息队列：消息队列本身可以实现简单的负载均衡，多个节点监听同一个队列，消息队列会将任务均匀分配给各个节点。
- 动态负载均衡：引入一个管理节点，定期收集各个工作节点的负载信息（如 CPU 使用率、内存使用率、当前处理任务数等），根据这些信息动态调整任务分配策略，将任务分配到负载较轻的节点。
故障恢复：
- 节点故障检测：管理节点定期向工作节点发送心跳包，工作节点回复心跳响应。如果管理节点在一定时间内未收到某个工作节点的心跳响应，则判定该节点故障。
- 任务重分配：当检测到某个节点故障时，管理节点将该节点上未处理完的任务重新分配到其他正常工作的节点。

关键代码示例

任务处理流水线示例

package main

import (
    "fmt"
)

// 定义任务结构体
type Task struct {
    ID int
    Data string
}

// 第一个处理阶段
func stage1(tasks <-chan Task, output chan<- Task) {
    for task := range tasks {
        // 模拟任务处理
        task.Data = "Processed by stage1: " + task.Data
        output <- task
    }
    close(output)
}

// 第二个处理阶段
func stage2(tasks <-chan Task, output chan<- Task) {
    for task := range tasks {
        // 模拟任务处理
        task.Data = "Processed by stage2: " + task.Data
        output <- task
    }
    close(output)
}

func main() {
    tasks := make(chan Task)
    stage1Output := make(chan Task)
    finalOutput := make(chan Task)

    // 启动流水线阶段
    go stage1(tasks, stage1Output)
    go stage2(stage1Output, finalOutput)

    // 提交任务
    for i := 0; i < 5; i++ {
        tasks <- Task{ID: i, Data: fmt.Sprintf("Task %d data", i)}
    }
    close(tasks)

    // 获取最终结果
    for result := range finalOutput {
        fmt.Println(result)
    }
}

简单的节点通信与负载均衡示例（基于 NATS）

package main

import (
    "fmt"

    "github.com/nats-io/nats.go"
)

func worker(nc *nats.Conn, workerID int) {
    sub, err := nc.Subscribe("tasks", func(m *nats.Msg) {
        // 处理任务
        fmt.Printf("Worker %d received task: %s\n", workerID, string(m.Data))
        // 处理完成后发送结果
        nc.Publish("results", []byte("Task processed by worker "+fmt.Sprintf("%d", workerID)))
    })
    if err != nil {
        fmt.Println("Subscription error:", err)
        return
    }
    defer sub.Unsubscribe()

    select {}
}

func main() {
    nc, err := nats.Connect(nats.DefaultURL)
    if err != nil {
        fmt.Println("NATS connection error:", err)
        return
    }
    defer nc.Close()

    // 启动多个工作节点
    for i := 0; i < 3; i++ {
        go worker(nc, i)
    }

    // 发送任务
    for j := 0; j < 5; j++ {
        err := nc.Publish("tasks", []byte(fmt.Sprintf("Task %d", j)))
        if err != nil {
            fmt.Println("Task publish error:", err)
        }
    }

    // 接收结果
    sub, err := nc.Subscribe("results", func(m *nats.Msg) {
        fmt.Println("Received result:", string(m.Data))
    })
    if err != nil {
        fmt.Println("Result subscription error:", err)
        return
    }
    defer sub.Unsubscribe()

    select {}
}

简单的故障检测与任务重分配示例

package main

import (
    "fmt"
    "sync"
    "time"
)

type Worker struct {
    ID int
    Alive bool
    Tasks []Task
}

type Task struct {
    ID int
    Data string
}

var (
    workers = make([]*Worker, 0)
    mutex   sync.Mutex
)

func heartbeat(worker *Worker) {
    for {
        // 模拟心跳
        fmt.Printf("Worker %d is alive\n", worker.ID)
        time.Sleep(3 * time.Second)
    }
}

func manager() {
    for {
        mutex.Lock()
        for _, worker := range workers {
            // 模拟心跳检测失败
            if!worker.Alive {
                // 重新分配任务
                for _, task := range worker.Tasks {
                    // 这里可以实现将任务分配到其他节点的逻辑
                    fmt.Printf("Re - assigning task %d from worker %d\n", task.ID, worker.ID)
                }
                // 从工作节点列表中移除故障节点
                for i, w := range workers {
                    if w.ID == worker.ID {
                        workers = append(workers[:i], workers[i+1:]...)
                        break
                    }
                }
            }
        }
        mutex.Unlock()
        time.Sleep(5 * time.Second)
    }
}

func main() {
    // 初始化工作节点
    for i := 0; i < 3; i++ {
        worker := &Worker{ID: i, Alive: true}
        workers = append(workers, worker)
        go heartbeat(worker)
    }

    go manager()

    select {}
}

星途面试题库

面试题：Go 并发设计模式在分布式系统中的应用优化

知识考点

面试题答案

设计思路

关键代码示例