Golang中Map、互斥锁与协程的实战应用

确保在所有这些映射访问周围包含足够的同步，以在同一同步操作中捕获测试和设置。这正是容易出错并导致竞态条件的地方。

这种底层同步的复杂性正是第一条 Go 谚语（https://go-proverbs.github.io/）的灵感来源。Go 确实提供了必要的原语，但更符合惯用法的是使用通道通信。

最后一次编辑，我保证。如果你以前没有做过这种级别的异步编码，那么值得在中等复杂的情况下多做几次，以便你能真正体会到这条谚语的含义。

我来回复这个问题，因为我不太确定你什么时候能有更多时间 😊

我有一个需要由多个goroutine完成的“任务”映射。所有goroutine必须完成所有任务，但同一时间不能有超过一个goroutine运行同一个任务。

你的意思是，假设你有3个goroutine，我称之为a、b和c，所有这3个goroutine都必须处理每个任务（例如，任务1需要a、b和c在其上执行，任务2也需要a、b和c在其上执行，等等）？任务的顺序重要吗（即a需要在b之前完成，b需要在c之前完成吗）？

如果是这样，我认为你只需要这样做：

type task struct { /* ... */ }

type taskResult struct {
    key string
    task task
}

results := make(chan taskResult, 8) // 根据你需要的缓冲大小设置
wg := sync.WaitGroup{}

// 从你的“主”goroutine遍历键和值；不需要加锁
for k, v := range tasks {
    wg.Add(1)
    go func(key string, task task) {
        defer wg.Done()
        // 你不需要在a、b和c之间进行任何同步，因为
        // 每个worker按顺序执行它们，但每个作业作为一个整体
        // 仍然是并发完成的
        doA(key, task)
        doB(key, task)
        doC(key, task)
        results <- taskResult{key, task}
    }(k, v)
}
go func() {
    wg.Wait()
    close(results)
}()
for _, res := range results {
    // 对结果进行处理。也许你需要更新映射？
    // 不过，你不需要任何锁，因为映射没有被并发访问。
    tasks[res.key] = res.task
}

这是一个典型的并发任务调度问题，需要确保任务互斥执行的同时避免全局锁的瓶颈。以下是几种解决方案：

方案1：使用sync.Map + sync.Mutex组合

package main

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

type TaskManager struct {
    tasks     sync.Map  // string -> *sync.Mutex
    completed sync.Map  // 记录已完成的任务
}

func (tm *TaskManager) DoTask(taskID string, workerID int) {
    // 获取或创建任务的互斥锁
    mu, _ := tm.tasks.LoadOrStore(taskID, &sync.Mutex{})
    taskMu := mu.(*sync.Mutex)
    
    // 锁定当前任务
    taskMu.Lock()
    defer taskMu.Unlock()
    
    // 检查任务是否已完成
    if _, ok := tm.completed.Load(taskID); ok {
        fmt.Printf("Worker %d: Task %s already completed\n", workerID, taskID)
        return
    }
    
    // 执行任务
    fmt.Printf("Worker %d: Starting task %s\n", workerID, taskID)
    time.Sleep(100 * time.Millisecond) // 模拟任务执行
    fmt.Printf("Worker %d: Completed task %s\n", workerID, taskID)
    
    // 标记任务完成
    tm.completed.Store(taskID, true)
}

func main() {
    tm := &TaskManager{}
    var wg sync.WaitGroup
    
    tasks := []string{"task1", "task2", "task3", "task1", "task2", "task3"}
    
    for i, task := range tasks {
        wg.Add(1)
        go func(workerID int, taskID string) {
            defer wg.Done()
            tm.DoTask(taskID, workerID)
        }(i, task)
    }
    
    wg.Wait()
}

方案2：使用通道进行任务调度

package main

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

type TaskScheduler struct {
    taskQueue chan string
    done      chan struct{}
    active    sync.Map // 正在执行的任务
}

func NewTaskScheduler(workers int) *TaskScheduler {
    ts := &TaskScheduler{
        taskQueue: make(chan string, 100),
        done:      make(chan struct{}),
    }
    
    // 启动worker池
    for i := 0; i < workers; i++ {
        go ts.worker(i)
    }
    
    return ts
}

func (ts *TaskScheduler) worker(id int) {
    for task := range ts.taskQueue {
        // 使用sync.Map确保任务互斥
        if _, loaded := ts.active.LoadOrStore(task, struct{}{}); loaded {
            // 任务已在执行，重新入队
            go func(t string) { ts.taskQueue <- t }(task)
            continue
        }
        
        // 执行任务
        fmt.Printf("Worker %d: Processing %s\n", id, task)
        time.Sleep(100 * time.Millisecond)
        fmt.Printf("Worker %d: Finished %s\n", id, task)
        
        // 移除活跃任务标记
        ts.active.Delete(task)
    }
}

func (ts *TaskScheduler) AddTask(task string) {
    ts.taskQueue <- task
}

func (ts *TaskScheduler) Wait() {
    close(ts.taskQueue)
    // 等待所有任务完成
    for len(ts.taskQueue) > 0 {
        time.Sleep(10 * time.Millisecond)
    }
}

func main() {
    scheduler := NewTaskScheduler(3)
    
    tasks := []string{"A", "B", "C", "A", "B", "C", "D", "E"}
    
    for _, task := range tasks {
        scheduler.AddTask(task)
    }
    
    scheduler.Wait()
}

方案3：使用单飞模式（SingleFlight）

package main

import (
    "fmt"
    "golang.org/x/sync/singleflight"
    "sync"
    "time"
)

type TaskExecutor struct {
    group singleflight.Group
    done  sync.Map
}

func (te *TaskExecutor) Execute(task string, workerID int) (interface{}, error) {
    // 检查是否已完成
    if _, ok := te.done.Load(task); ok {
        fmt.Printf("Worker %d: Task %s already done\n", workerID, task)
        return nil, nil
    }
    
    // 使用SingleFlight确保同一任务只执行一次
    result, err, _ := te.group.Do(task, func() (interface{}, error) {
        fmt.Printf("Worker %d: Executing task %s\n", workerID, task)
        time.Sleep(100 * time.Millisecond)
        
        // 标记完成
        te.done.Store(task, true)
        
        return fmt.Sprintf("Result of %s", task), nil
    })
    
    if err == nil {
        fmt.Printf("Worker %d: Got result for %s: %v\n", workerID, task, result)
    }
    
    return result, err
}

func main() {
    executor := &TaskExecutor{}
    var wg sync.WaitGroup
    
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            task := fmt.Sprintf("task%d", id%3) // 只有3个不同的任务
            executor.Execute(task, id)
        }(i)
    }
    
    wg.Wait()
}

方案4：改进的互斥锁映射方案

package main

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

type SafeTaskMap struct {
    mu    sync.RWMutex
    tasks map[string]*taskEntry
}

type taskEntry struct {
    mu       sync.Mutex
    done     bool
    waiters  []chan struct{}
}

func NewSafeTaskMap() *SafeTaskMap {
    return &SafeTaskMap{
        tasks: make(map[string]*taskEntry),
    }
}

func (stm *SafeTaskMap) Execute(taskID string, workerID int) {
    // 快速路径：只读锁检查
    stm.mu.RLock()
    entry, exists := stm.tasks[taskID]
    stm.mu.RUnlock()
    
    if !exists {
        // 需要创建新条目
        stm.mu.Lock()
        // 双重检查
        if entry, exists = stm.tasks[taskID]; !exists {
            entry = &taskEntry{}
            stm.tasks[taskID] = entry
        }
        stm.mu.Unlock()
    }
    
    // 锁定具体任务
    entry.mu.Lock()
    
    if entry.done {
        entry.mu.Unlock()
        fmt.Printf("Worker %d: Task %s already completed\n", workerID, taskID)
        return
    }
    
    // 执行任务
    fmt.Printf("Worker %d: Starting task %s\n", workerID, taskID)
    time.Sleep(100 * time.Millisecond)
    fmt.Printf("Worker %d: Completed task %s\n", workerID, taskID)
    
    entry.done = true
    entry.mu.Unlock()
}

func main() {
    stm := NewSafeTaskMap()
    var wg sync.WaitGroup
    
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            task := fmt.Sprintf("task%d", id%4)
            stm.Execute(task, id)
        }(i)
    }
    
    wg.Wait()
}

推荐方案1，它结合了sync.Map的并发安全特性和互斥锁的任务级保护，避免了全局锁的瓶颈。每个任务有自己的互斥锁，不同任务可以并行执行，相同任务会串行执行，且已完成的任务会快速返回。