掘金 后端 ( ) • 2024-05-04 12:40
Go使用tips: Concurrency & Synchronization 并发同步原语
Concurrency &amp Synchronization Tips 原语

highlight: a11y-dark theme: z-blue

翻译：https://github.com/func25/go-practical-tips/tree/main

优先选择 chan struct{} 而不是 chan bool 来处理各程序之间的信号传递

当我们使用 goroutines 并需要在它们之间发送信号时，我们可能会纠结是使用 chan bool 还是 chan struct{}。

为什么偏爱 “chan struct{}”？

那么，chan bool 也可以发出事件信号，对吗？它发送的是一个布尔值（true 或 false），根据设置的不同，可能会有一些特定的含义。但问题就在这里：

type JobDispatcher struct {
    start chan bool
}


func NewJobDispatcher() *JobDispatcher {
    return &JobDispatcher{
        start: make(chan bool),
    }
}

// Unclear: What does sending true or false mean?

现在，让我们来谈谈 struct{} 类型，这种类型纯粹是用来发出信号的，因为 struct{} 类型根本不占用任何内存，就像在说 “嘿，发生了什么事 ”一样： 就像在说：“嘿，发生了一些事情。”而不会发送任何实际数据。

type JobDispatcher struct {
    start chan struct{}
}

func NewJobDispatcher() *JobDispatcher {
    return &JobDispatcher{
        start: make(chan struct{}),
    }
}

func (j *JobDispatcher) Start() {
    j.start <- struct{}{}
}

// Clear: Sending anything means "start the job"

那么，它的主要优势是什么呢？

  • 首先，由于 struct{} 的大小为零，因此通过 struct{} 发送值实际上不会在通道上移动任何数据，而只是发送信号。这是一个微妙但不错的内存优化。
  • 如果开发人员在代码中看到 chan struct{}，就会立即明白这条channel是用于发送信号的，从而减少混淆。

缺点是什么？

虽然结构struct{}{}语法有些别扭，但这小小的不便也是值得的，因为它可以防止通道被滥用于传输数据，而你想要的只是一个简单的信号。

对于一次性信号，您甚至可能不需要发送数值。您可以直接关闭channel：


func (j *JobDispatcher) Start() {
    close(j.start)
}

关闭channel是一种明确而有效的方式，可以向多个接收器广播作业应开始的信号，而无需发送任何数据

Buffered Channels作为限制 Gooutine 执行的信号传递器

当我们希望控制有多少个 goroutines 可以同时访问特定资源时，使用信号是一个不错的选择。我们可以简单地使用 Go 中的缓冲通道创建一个信号。

Buffered Channels的大小决定了可以并发运行的 goroutines 的数量：

semaphore := make(chan struct{}, numTokens)

下面是基本流程：

  • 一个 goroutine 会尝试向通道发送一个值，占用一个可用槽。
  • 一旦例行程序完成任务，它就会从通道中移除数值，从而腾出空位给另一个例行程序使用。
var wg sync.WaitGroup
wg.Add(10)

for i := 0; i < 10; i++ {
    go func(id int) {
        defer wg.Done()
        semaphore <- struct{}{} // Acquire a token.
        ...
        <-semaphore // Release the token.
    }(i)
}

wg.Wait()

在此代码段中

  • wg.Add(10)：设置 10 个运行程序。
  • make(chan struct{}, 3)：初始化一个只允许 3 个运行程序同时运行的 Semaphore。 如果你正在寻找一种更有条理的方式，我们可以定义一个 Semaphore 类型来封装所有与 semaphore 相关的操作：
type Semaphore chan struct{}

func NewSemaphore(maxCount int) Semaphore {
	return make(chan struct{}, maxCount)
}

func (s Semaphore) Acquire() {
	s <- struct{}{}
}

func (s Semaphore) Release() {
	<-s
}

使用这种自定义 Semaphore 类型简化了我们管理资源访问的方式：

func doSomething(semaphore *Semaphore) {
    semaphore.Acquire()
    defer semaphore.Release()

    ...
}

此外，对于更复杂的情况，您可能需要了解 http://golang.org/x/sync/semaphore 软件包，它提供了一种加权信号实现。

当某些任务可能比其他任务需要更多资源时，这一点尤其有用，例如管理数据库连接池，其中某些操作需要同时使用多个连接。

加权 semaphores 允许单个 goroutine 同时占用多个插槽。

使用 singleflight 优化多次调用

比方说，我们有一个从某处提取数据的函数，它并不快，每次调用大约需要 3 秒钟：

func FetchExpensiveData() (int64, error) {
    time.Sleep(3 * time.Second)

    return time.Now().Unix() / 10, nil
}

模拟函数会在 10 秒后发出一个不同的数字。

现在，如果我们连续调用这个函数 3 次，我们需要等待大约 9 秒钟。

我们可能会认为使用 3 个 goroutines 可以将等待时间缩短到 3 秒左右，但我们仍然要运行 3 次函数，而且都是为了得到相同的结果。

这就是 singleflight 软件包真正能改变游戏规则的地方，它能确保无论我们在这 3 秒钟内调用函数多少次，它实际上只运行一次，并向每个调用者返回相同的结果。

我们可以在 http://golang.org/x/sync/singleflight 找到它。

现在，我们来看看如何使用它：

var group singleflight.Group

func UsingSingleFlight(key string) {
    v, _, _ = group.Do(key, func() (any, error) {
        return FetchExpensiveData()
    })

    fmt.Println(v)
}

这里发生的事情非常简单。

我们先创建一个 singleflight.Group，然后在 group.Do() 方法中封装昂贵的函数调用，这个方法很聪明。

它会检查是否已经请求过相同的密钥。如果是，它会等待原始调用的结果，并将其返回给所有调用者，而不仅仅是第一个调用者。

基本上就是这样。

“key ”参数的作用是什么？

它告诉 singleflight，多个请求实际上是在请求同一件事。因此，它会使用这个键来检查是否应该再次运行函数，或者只是等待并返回正在进行的操作的结果。

如果你想看看实际代码是如何实现的，可以点击这里查看： https://go.dev/play/p/30kdFPsy2HR

基本上，如果同一函数被同时调用多次，则只进行一次真正的调用，然后由所有调用者共享这一次调用的结果。

“为什么不使用缓存？

Singleflight 并不是缓存，而是一种获取结果或调用可同时运行的函数的方法。

该函数可能不会返回任何结果，这个例子可能会让人误解这是为了处理昂贵的数据，但我们只是想确保只调用一次该函数。

就像向服务器发送 ping() 函数，你不会缓存 ping 的结果，但也不想让服务器同时承受多个 ping 的重负。

sync.Once 是只做一次事情的最佳方法

我们知道，有时我们需要确保某些事情在应用程序中只发生一次，即使有大量事情在同时运行。

让我们用 sync.Once 来谈谈这个问题，它在设置单例时非常常见。

例如，我们有一个配置对象，对吗？我们需要对它进行一次设置，然后在应用程序的其他地方使用相同的设置：

var instance *Config

func GetConfig() *Config {
    if instance == nil {
        instance = loadConfig()
    }

    return instance
}

但问题是，如果有很多 goroutines 同时尝试获取配置，而配置尚未设置好，那么最终可能会多次运行 loadConfig()，这可不是我们想要的结果。

这就是 sync.Once 的作用所在：

var (
    once     sync.Once
    instance *Config
)

func GetConfig() *Config {
    once.Do(func() {
        instance = loadConfig()
    })

    return instance
}

我们将设置代码封装在一个函数中，并将其传递给 once.Do()，这样，无论有多少个程序同时调用 GetConfig()，loadConfig() 函数都能保证只运行一次。

这样就能确保每个人都能获得相同的 Config 实例，而无需多次设置。

现在，重要的是要明白，sync.Once 并不意味着函数在一般意义上只运行一次，而是只在该特定 sync.Once 实例的上下文中执行一次，这是一个很大的区别。

问题是，如果你尝试在不同的函数中再次使用 sync.Once，即使新函数之前没有使用过 sync.Once，它也不会起作用：

o.Do(f1)
o.Do(f2)

如果 f1 已经运行，第二个函数 f2 就会被忽略。这是因为 sync.Once 并不是关于函数，而是关于只做一次的行为，而不管那是什么。

从 Go 1.21 版本开始，sync 包提供了更多有用的功能。我们现在有了

sync.OnceFunc、sync.OnceValue 和 sync.OnceValues。

这些新函数将标准函数转换为 sync.Once 类函数，确保无论函数被调用多少次，都只执行一次。

func OnceFunc(f func()) func()
func OnceValue[T any](f func() T) func() T
func OnceValues[T1, T2 any](f func() (T1, T2)) func() (T1, T2)

当您需要从一个只执行一次的函数中返回值时，甚至包括处理错误时，这一新增功能尤其有用：

var GetConfig = sync.OnceValues(func() (Config, error) {
    return Config{..}, nil
})

GetConfig 只执行一次以检索配置，可能包括错误，第一次调用后，任何对 GetConfig 的进一步调用都将返回与第一次相同的结果，函数本身实际上不会再次运行。

让我们深入了解一下 sync.Once 的内部工作原理：

  • 一个原子计数器，可以是 0 或 1。
  • 一个互斥器，用于保护较慢的操作。 现在，让我为你分析一下 sync.Once 的结构：
type Once struct {
    done atomic.Uint32
    m    Mutex
}

快速路径

调用 once.Do(f) 时，首先会检查原子计数器，如果计数器为 0，则表示函数尚未运行。

如果函数已经执行完毕，则后续调用可以绕过该检查，从而大大加快速度。

慢速路径

现在，如果计数器为 0，sync.Once 会切换到所谓的慢速路径，即执行函数：

func(o *Once) doSlow(f func()) {
    o.m.Lock()
    defer o.m.Unlock()

    if o.done.Load() == 0 {
        defer o.done.Store(1)
        f()
    }
}
  • o.m.Lock()： 这将锁定互斥项，以确保在任何时候只有一个 goroutine 可以执行下面的步骤。
  • o.done.Load()==0：一旦锁定，它会再次检查计数器，以确保在此期间没有其他 goroutine 运行该函数。
  • o.done.Store(1)： 函数执行完毕后，它将计数器更新为 1，表明函数已被运行，不应再次运行。

为什么会有快速路径和慢速路径

我们之所以有快速路径和慢速路径，是为了在速度和安全性之间取得平衡。

快速路径允许系统绕过互斥锁，如果函数已经执行，则绕过互斥锁执行，这样速度会快很多。

而慢速路径则能确保函数第一次执行时的安全性，不会受到其他程序的干扰。这种初始设置可能会慢一些，但一旦完成，每次调用 once.Do() 都会很快，这从长远来看是件好事。

使用 errgroup 管理Goroutines

当我们处理多个 goroutines 时，管理它们及其错误会变得有点棘手。

我们可能已经熟悉了用于处理多个 goroutines 的 sync.WaitGroup。但是，还有一个名为 errgroup 的工具，它提供了一些巧妙的功能，可以进一步简化这一过程。

首先，你需要下载包：

$ go get -u golang.org/x/sync

想象一下，您想同时从多个 URL 获取数据：

func main() {
    urls := []string {
        "https://blog.devtrovert.com",
        "https://example.com",
    }

    var g errgroup.Group

    for _, url := range urls {
        url := url // safe before Go 1.22
        g.Go(func() error {
            return fetch(url)
        })
    }

    if err := g.Wait() ; err != nil {
        log.Fatal(err)
    }
}

在这个代码段中，我们同时获取 2 个页面。g.Wait() 方法是我们要重点关注的方法，它会等待所有 goroutines 结束。如果任何一个 goroutines 遇到错误，g.Wait() 将返回它遇到的第一个错误。

Errgroup 简化了管理多个 goroutines 和处理执行过程中可能出现的错误的过程。

让我们来分析一下关键点：

  • 使用 g.Go()，在自己的 goroutine 中启动每个任务，并将执行工作的函数传递给它。
  • 使用 g.Wait()等待所有 goroutine 完成，它会返回 goroutine 中遇到的第一个错误，而不是所有错误。
  • Errgroup 可与上下文无缝连接。通过使用 errgroup.WithContext()，如果任何 goroutine 失败并返回错误，上下文将自动取消。

内部结构

Errgroup.Group 结构使用 Go 标准库中的一些组件构建：

type Group struct {
    cancel func()
    
    wg sync.WaitGroup
    sema chan struct{}
    errOnce sync.Once

    err error
}
  • wg sync.WaitGroup： 用于等待所有程序完成任务。
  • errOnce sync.Once：确保以线程安全的方式捕获第一个错误，即在设置错误时不会出现竞赛条件。
  • sema chan struct{}： 一个 semaphore 可以控制同时运行的 goroutines 数量。我们甚至可以使用 errg.SetLimit() 设置并发程序的数量限制。

无论何时使用 g.Go()，本质上都是在向组中添加一个新任务。该任务是一个不带任何参数但会返回错误的函数。可以同时运行的并发 goroutines 的数量由一个 semaphore 来管理，它有助于控制执行并防止过多的任务同时运行。

下面是具体的实现过程：

func (g *Group) Go(f func() error) {
	if g.sem != nil {
		g.sem <- token{}
	}

	g.wg.Add(1)
	go func() {
		defer g.done()

		if err := f(); err != nil {
			g.errOnce.Do(func() {
				g.err = err
				if g.cancel != nil {
					g.cancel(g.err)
				}
			})
		}
	}()
}

无论何时使用 g.Go()，本质上都是在向组中添加一个新任务。该任务是一个不带任何参数但会返回错误的函数。可以同时运行的并发 goroutines 的数量由一个 semaphore 来管理，它有助于控制执行并防止过多的任务同时运行。

下面是具体的实现过程：

func (g *Group) done() {
    if g.sem != nil {
        <-g.sem
    }
    g.wg.done()
}

func (g *Group) Wait() error {
	g.wg.Wait()
	if g.cancel != nil {
		g.cancel(g.err)
	}
	return g.err
}

函数 done() 起着重要作用，它不仅向 WaitGroup 发送 goroutine 已完成的信号，还管理着 semaphore，以确保运行 goroutine 的限制得到遵守。

我们还应该记住，使用 goroutines 并不总是最好的解决方案，尤其是对于快速完成的任务。有时，按顺序一个接一个地运行任务可能会更快，也更容易管理。

安全的泛型sync.Pool

那么，让我们来看看 sync.Pool。对于那些可能不太熟悉的人来说，sync.Pool 是 Go 标准库中的一项功能，主要用于重复使用对象。

这非常方便，因为它有助于减少内存分配的次数，从而大大提高性能。

想象一下，你有一台超快的打印机，每分钟需要打印 100 页。如果打印机每打印一页都要跑到储藏室去拿新的墨水和纸张，那就没什么意义了，对吗？

相反，你可以这样想：打印机有一个装有大约 100 张纸的托盘，可以重复使用。你手头有一批固定的物品，可以反复使用，这样既节省了时间，又节省了资源。

下面是代码中的一个小片段：

var paperPool = sync.Pool{
    New: func() interface{} {
        return new(Paper) // create a new sheet of paper
    }
}

func printPage() {
    page := paperPool.Get().(*Paper)
    page.Reset() // make sure to reset the page before using it

    defer paperPool.Put(page)

    page.Print()
}

但有几点需要注意：

  • sync.Pool 没有固定大小，这意味着你可以不断添加和检索项目，而不受任何硬性限制。
  • 一旦你将一个对象放回池中，就不要再想它了，因为它可能会被删除或被垃圾回收器回收。
  • 由于对象可能有状态，因此在将其放回池中或取出后，清除或重置其状态至关重要。

泛型池

我们已经知道 sync.Pool 使用空接口{}来存储和检索项目。这非常灵活，但不提供类型安全，我们可以对这个过程进行包装，使其类型安全：


type Pool[T any] struct {
    internal sync.Pool
}

func NewPool[T any](newF func() T) *Pool[T] {
    return &Pool[T]{
        internal: &sync.Pool{
            New: func() interface{} {
                return newF()
            }
        }
    }
}

通过这种设置，我们创建了一个与特定类型 T 绑定的池，尽管它在引擎盖下仍使用 interface{}。神奇之处在于我们如何以类型安全的方式处理数据的获取和放置：

func (p *Pool[T]) Get() T {
    return p.internal.Get().(T)
}

func (p *Pool[T]) Put(x T) {
    p.internal.Put(x)
}

从池中获取数据时，我们只需将接口转换为 T 类型。你可能会问，为什么我们不在转换过程中检查错误呢？

“为什么我们不在转换过程中检查错误呢？

事情是这样的：我们添加的泛型层为我们确保了类型，因此无需担心转换失败。

sync.Pool 永远只包含 T 类型的实例，因此断言 p.internal.Get().(T) 是安全的，在正常情况下不会引起panic。这种解决方案使我们的代码简洁高效。

sync.Mutex Embedding 互斥锁嵌入

在应用程序的不同部分需要并发访问共享资源时，通常会使用 sync.Mutex 来管理访问并确保数据完整性。

通常情况下，您可能会看到类似这样的情况：

type MyStruct struct {
    mu sync.Mutex

    ...
}

func (s *MyStruct) DoSomething() {
    s.mu.Lock()
    defer s.mu.Unlock()

    ...
}

这种模式虽然有效，但会导致代码中充满 mu.Lock() 和 mu.Unlock() 调用，从而增加代码的阅读和维护难度。

一种更简洁的处理方法是将 sync.Mutex 直接嵌入到结构体中，这样就可以直接在结构体的实例上调用 Lock 和 Unlock：


type Lockable[T any] struct {
    sync.Mutex
    Value T
}

func (l *Lockable[T]) SetValue(v T) {
    l.Lock()
    defer l.Unlock()

    l.Value = v
}

func (l *Lockable[T]) GetValue() T {
    l.Lock()
    defer l.Unlock()

    return l.Value
}

为什么我们不直接嵌入 T 类型，而要使用 Value T？

原因是 Go 目前不支持在结构体中直接嵌入类型参数。

使用channel 不被阻塞

通常情况下，当我们通过channel发送信息时，代码会坐在那里等待接收器准备好接收数值，就像这样

ch <- value // blocking operation

但如果我们遇到了不想等待的情况呢？

也许我们已经有了一个使用信号传递器管理资源的系统，而且我们已经有了一个像 TryAcquire() 这样的函数，它应该立即告诉我们是否因为资源已被使用而无法获取资源。

让我们以 errgroup 软件包为例，看看在实践中是如何处理类似问题的。该软件包内部使用了一种简单的信号机制来限制同时运行的 goroutines 数量。

下面是如果信号灯已满，而你又试图启动一个新的 goroutine 时会发生的情况：

func (g *Group) TryGo(f func() error) bool {
    if g.sem != nil {
        select {
        case g.sem <- token{}:
        default:
            return false
        }
    }

    ...
}

神奇之处在于 select{} 语句。通常，select{} 用于等待多个通道操作，但在这里，它被巧妙地用于尝试非阻塞发送：

  • case g.sem <- token{}： 这一行尝试向 semaphore 通道 g.sem 发送令牌。如果通道中还有空间（这意味着它没有达到容量），则标记发送成功，函数继续执行。
  • default： 如果 g.sem 通道已经满了，这部分就会启动。它不会阻塞并等待空间释放，而是直接进入默认情况。

选择语句中的默认情况是一个棘手的路径，因为如果没有其他情况就绪，它就会立即执行。在这种情况下，它会立即返回 false，让我们知道我们无法启动新的 goroutine，因为我们已经达到了为活动 goroutine 设置的限制。