分支、循环传输、接收者数量不确定

从现在开始，努力学习吧！本文《分支、循环传输、接收者数量不确定》主要讲解了等等相关知识点，我会在golang学习网中持续更新相关的系列文章，欢迎大家关注并积极留言建议。下面就先一起来看一下本篇正文内容吧，希望能帮到你！

我需要一种方法来多次从一个主 goroutine 和未知数量的其他 goroutine 发出信号。我还需要那些其他 goroutine 在多个项目上 select ，所以忙等待（可能）不是一个选择。我想出了以下解决方案：

package main

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

type signal struct {
    data     []int
    channels []chan struct{}
}

func newSignal() *signal {
    s := &signal{
        data:     make([]int, 0),
        channels: make([]chan struct{}, 1),
    }
    s.channels[0] = make(chan struct{})
    return s
}

func (s *signal) Broadcast(d int) {
    s.data = append(s.data, d)
    s.channels = append(s.channels, make(chan struct{}))
    close(s.channels[len(s.data)-1])
}

func test(s *signal, wg *sync.WaitGroup, id int, ctx context.Context) {
    for i := 0; ; i += 1 {
        select {
        case <-s.channels[i]:
            if id >= s.data[i] {
                fmt.Println("Goroutine completed:", id)
                wg.Done()
                return
            }
        case <-ctx.Done():
            fmt.Println("Goroutine completed:", id)
            wg.Done()
            return
        }
    }
}

func main() {
    s := newSignal()

    ctx, cancel := context.WithCancel(context.Background())
    wg := sync.WaitGroup{}
    wg.Add(3)
    go test(s, &wg, 3, ctx)
    go test(s, &wg, 2, ctx)
    go test(s, &wg, 1, ctx)

    s.Broadcast(3)
    time.Sleep(1 * time.Second)

    // multiple broadcasts is mandatory
    s.Broadcast(2)
    time.Sleep(1 * time.Second)

    // last goroutine
    cancel()

    wg.Wait()
}

演示：https://play.golang.org/p/dgmlktuj7ty

有没有更优雅的方法来做到这一点？一种仅使用内置库的方法。如果不是，这是一个安全/可以使用的解决方案吗？我相信它至少是安全的，因为它适用于大量的 goroutine（我已经用它做了一些测试）。

简而言之，这正是我想要的：

• 主 goroutine（称为 m）必须能够多次向一些未知数量的其他 goroutine（将其称为 n for 0...n）发送一些数据（称为 d），每个 goroutine 都需要一个每次基于 d 的操作
• m 必须能够多次向所有其他 n goroutines 发送特定（数字）数据的信号
• n 中的每个 goroutine 要么自行终止（基于上下文），要么在使用 d 执行某些操作并决定其命运后终止。它将按照发出的信号多次执行此检查，直到死亡。
• 我不允许以任何方式跟踪 n goroutine（例如，拥有 goroutine 的通道映射并进行迭代）

在我的解决方案中，通道切片不代表 goroutine：它们实际上代表正在广播出去的信号。这意味着，如果我广播两次，然后 goroutine 旋转起来，它将在 select 块中休眠之前检查两个信号。

解决方案

在我看来，您可能想要类似扇出模式的东西。下面的 source 描述了扇入和扇出以及其他并发模式。 golang.org 上的 blog post 也与此相关。我认为它本质上是 observer pattern 使用频道的版本。

基本上，您想要一些东西，比如 Broadcaster，它可以保留频道列表。当您调用 Broadcaster.send(data) 时，它会循环遍历在每个通道上发送 data 的通道列表。 Broadcaster 还必须有一种方法让 goroutine 将 订阅 到 Broadcaster。 Goroutines 必须有办法接受来自 Broadcaster 的通道，或者向 Broadcaster 提供通道。该通道就是通信链路。

如果要在“观察者”goroutine 中执行的工作需要很长时间，请考虑使用缓冲通道，以便 Broadcaster 在 send 和等待 goroutine 期间不会阻塞。如果你不关心 goroutine 是否错过了 data，你可以使用非阻塞发送（见下文）。

当一个 Goroutine “死亡”时，它可以从 Broadcaster 订阅 unsubscribe ，这将从其列表中删除相应的通道。或者通道可以保持满，并且 Broadcaster 将必须使用 non-blocking send 跳过完整通道到死亡 goroutine。

我不能说我所描述的内容是全面的或100%正确的。这只是我根据您的问题陈述尝试的第一个一般性操作的快速描述。

文中关于的知识介绍，希望对你的学习有所帮助！若是受益匪浅，那就动动鼠标收藏这篇《分支、循环传输、接收者数量不确定》文章吧，也可关注golang学习网公众号了解相关技术文章。