17 | SingleFlight 和 CyclicBarrier：请求合并和循环栅栏该怎么用？

在并发编程的世界里，高效地管理多个线程或协程的协作与同步是确保系统稳定性和性能的关键。Go语言（Golang）以其简洁的语法和强大的并发模型，特别是goroutines和channels，成为了处理并发任务的首选语言之一。然而，在复杂的应用场景中，仅仅依靠goroutines和channels可能不足以解决所有问题。本章将深入探讨Go标准库中的两个高级同步工具：SingleFlight和CyclicBarrier（尽管CyclicBarrier并非直接包含在Go标准库中，但我们将通过自定义实现来探讨其应用），以及它们如何帮助我们在并发编程中实现请求合并和循环同步。

17.1 SingleFlight：避免重复请求

在Web服务或分布式系统中，经常需要处理来自多个客户端的相同请求。如果这些请求被独立处理，不仅会增加服务器的负载，还可能导致数据不一致或浪费资源。SingleFlight模式旨在解决这一问题，它通过确保对同一请求的并发访问只触发一次实际的处理过程，并将结果缓存起来供后续请求使用，从而避免重复工作。

17.1.1 SingleFlight 的基本原理

SingleFlight模式的核心在于维护一个全局的“请求-结果”映射表，以及一个用于协调请求处理的机制。当一个请求到达时，系统会检查该请求是否已经在处理中或已有处理结果。如果是，则直接返回已缓存的结果；否则，启动一个新的处理过程，并将结果存储在映射表中供后续使用。

17.1.2 Go标准库中的SingleFlight

Go标准库中的groupcache包提供了一个singleflight子包，实现了SingleFlight模式。这里是一个简单的使用示例：

package main
import (
    "fmt"
    "golang.org/x/sync/singleflight"
    "time"
)
var (
    group = &singleflight.Group{}
)
func fetchData(key string) (string, error) {
    // 模拟数据获取过程
    time.Sleep(2 * time.Second)
    return fmt.Sprintf("Data for %s", key), nil
}
func getData(key string) (string, error) {
    res, err, shared := group.Do(key, func() (interface{}, error) {
        return fetchData(key)
    })
    if err != nil {
        return "", err
    }
    return res.(string), nil
}
func main() {
    start := time.Now()
    go func() {
        fmt.Println(getData("key1"))
    }()
    go func() {
        fmt.Println(getData("key1"))
    }()
    time.Sleep(3 * time.Second) // 确保goroutines有足够时间执行
    fmt.Printf("Total time: %v\n", time.Since(start))
}

在上述代码中，尽管我们同时发起了两个对相同键"key1"的请求，但fetchData函数只会被调用一次，因为singleflight.Group确保了请求合并。

17.2 CyclicBarrier：循环栅栏同步

在并发任务中，有时需要等待一组线程都到达某个执行点后，再继续执行后续操作。这种场景在并行计算、游戏开发或任何需要精确控制多个任务同步执行的场景中尤为常见。CyclicBarrier是一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点（common barrier point）。

17.2.1 CyclicBarrier 的基本原理

CyclicBarrier的工作原理类似于一个计数器，它初始化时设定一个屏障点必须达到的线程数量（parties）。每当一个线程到达屏障点时，它会被阻塞，直到所有线程都到达。然后，所有线程被同时释放，继续执行后续操作。这个过程可以重复进行，因此得名“循环”栅栏。

17.2.2 Go中实现CyclicBarrier

由于Go标准库中没有直接提供CyclicBarrier的实现，我们可以使用goroutines、channels和sync.WaitGroup来自定义一个。

package main
import (
    "fmt"
    "sync"
    "time"
)
type CyclicBarrier struct {
    mu          sync.Mutex
    parties     int
    generation  int
    arrived     int
    resetChan   chan struct{}
    continueChan chan struct{}
}
func NewCyclicBarrier(parties int) *CyclicBarrier {
    return &CyclicBarrier{
        parties:     parties,
        generation:  0,
        arrived:     0,
        resetChan:   make(chan struct{}),
        continueChan: make(chan struct{}),
    }
}
func (cb *CyclicBarrier) Await() {
    cb.mu.Lock()
    gen := cb.generation
    cb.arrived++
    if cb.arrived < cb.parties {
        // 等待其他线程到达
        cb.mu.Unlock()
        <-cb.continueChan
    } else {
        // 所有线程都已到达，重置并通知
        nextGen := gen + 1
        close(cb.resetChan)
        cb.generation = nextGen
        cb.arrived = 0
        cb.continueChan = make(chan struct{})
        // 唤醒所有等待的线程
        close(cb.continueChan)
    }
    cb.mu.Unlock()
    // 如果是最后一个到达的线程，则等待重置信号
    if cb.arrived == cb.parties {
        <-cb.resetChan
    }
}
func main() {
    var wg sync.WaitGroup
    barrier := NewCyclicBarrier(3)
    for i := 0; i < 3; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            fmt.Printf("Thread %d ready\n", id)
            barrier.Await()
            fmt.Printf("Thread %d passed barrier\n", id)
        }(i + 1)
    }
    wg.Wait()
}

注意：上述CyclicBarrier实现为了简化理解，并未处理所有可能的并发错误情况，如竞态条件或死锁。在实际应用中，可能需要更复杂的锁策略和错误处理逻辑。

17.3 结合应用

SingleFlight和CyclicBarrier在并发编程中各有其用武之地，但它们也可以结合使用，以解决更复杂的问题。例如，在一个分布式系统中，多个节点可能同时请求相同的计算密集型任务。使用SingleFlight可以避免任务重复执行，而CyclicBarrier（或类似机制）可以确保在任务执行完毕后，所有节点能够同步地进入下一个阶段，比如更新缓存或响应客户端。

通过合理设计，我们可以将这两种同步机制融入到系统的各个层面，从而提升系统的整体性能和稳定性。

结语

SingleFlight和CyclicBarrier是并发编程中两个强大的工具，它们分别解决了请求合并和循环同步的问题。虽然Go标准库直接提供了SingleFlight的实现，但通过自定义我们可以模拟出CyclicBarrier的功能。在实际应用中，根据具体需求灵活选择和使用这些工具，将有助于我们构建更加高效、稳定的并发系统。