141

要开始执行两个 goroutine 的无限循环，我可以使用下面的代码：

收到 msg 后，它将启动一个新的 goroutine 并永远继续下去。

c1 := make(chan string)
c2 := make(chan string)

go DoStuff(c1, 5)
go DoStuff(c2, 2)

for ; true;  {
    select {
    case msg1 := <-c1:
        fmt.Println("received ", msg1)
        go DoStuff(c1, 1)
    case msg2 := <-c2:
        fmt.Println("received ", msg2)
        go DoStuff(c2, 9)
    }
}

我现在希望 N 个 goroutine 具有相同的行为，但是在这种情况下 select 语句会如何？

这是我开始使用的代码位，但我很困惑如何编写 select 语句

numChans := 2

//I keep the channels in this slice, and want to "loop" over them in the select statemnt
var chans = [] chan string{}

for i:=0;i<numChans;i++{
    tmp := make(chan string);
    chans = append(chans, tmp);
    go DoStuff(tmp, i + 1)

//How shall the select statment be coded for this case?  
for ; true;  {
    select {
    case msg1 := <-c1:
        fmt.Println("received ", msg1)
        go DoStuff(c1, 1)
    case msg2 := <-c2:
        fmt.Println("received ", msg2)
        go DoStuff(c2, 9)
    }
}
4

6 回答 6

169

您可以使用反射包中的Select函数执行此操作：

func Select(cases []SelectCase) (chosen int, recv Value, recvOK bool)

Select 执行由案例列表描述的选择操作。与 Go 的 select 语句一样，它会阻塞直到至少有一个案例可以继续，做出统一的伪随机选择，然后执行该案例。它返回所选案例的索引，如果该案例是接收操作，则返回接收到的值和一个布尔值，指示该值是否对应于通道上的发送（而不是因为通道关闭而接收到的零值）。

您传入一个结构数组，这些SelectCase结构标识要选择的通道、操作的方向以及在发送操作的情况下要发送的值。

所以你可以做这样的事情：

cases := make([]reflect.SelectCase, len(chans))
for i, ch := range chans {
    cases[i] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ch)}
}
chosen, value, ok := reflect.Select(cases)
// ok will be true if the channel has not been closed.
ch := chans[chosen]
msg := value.String()

您可以在这里尝试一个更充实的示例：http ://play.golang.org/p/8zwvSk4kjx

于 2013-11-15T02:37:56.830 回答
110

您可以通过将每个通道包装在一个 goroutine 中来实现这一点，该 goroutine 将消息“转发”到共享的“聚合”通道。例如：

agg := make(chan string)
for _, ch := range chans {
  go func(c chan string) {
    for msg := range c {
      agg <- msg
    }
  }(ch)
}

select {
case msg <- agg:
    fmt.Println("received ", msg)
}

如果您需要知道消息来自哪个通道，您可以在将其转发到聚合通道之前将其包装在带有任何额外信息的结构中。

在我的（有限）测试中，这种方法使用反射包的性能大大优于：

$ go test dynamic_select_test.go -test.bench=.
...
BenchmarkReflectSelect         1    5265109013 ns/op
BenchmarkGoSelect             20      81911344 ns/op
ok      command-line-arguments  9.463s

基准代码在这里

于 2015-09-02T00:08:00.993 回答
28

为了扩展对先前答案的一些评论并提供更清晰的比较，这里给出了迄今为止提出的两种方法的示例，给出相同的输入、要读取的通道切片和调用每个值的函数，这些值还需要知道哪个渠道价值来自。

这些方法之间存在三个主要区别：

  • 复杂。虽然它可能部分是读者偏好，但我发现渠道方法更惯用、直接且易读。

  • 表现。在我的 Xeon amd64 系统上，goroutines+channels out 执行反射解决方案大约两个数量级（一般来说，Go 中的反射通常较慢，只应在绝对需要时使用）。当然，如果函数处理结果或将值写入输入通道时有任何显着延迟，则这种性能差异很容易变得微不足道。

  • 阻塞/缓冲语义。这一点的重要性取决于用例。大多数情况下，这无关紧要，或者 goroutine 合并解决方案中的轻微额外缓冲可能有助于吞吐量。但是，如果希望只有一个写入器被解除阻塞并且它的值在任何其他写入器被解除阻塞之前被完全处理，那么这只能通过反射解决方案来实现。

请注意，如果不需要发送通道的“id”或者源通道永远不会关闭，这两种方法都可以简化。

Goroutine 合并通道：

// Process1 calls `fn` for each value received from any of the `chans`
// channels. The arguments to `fn` are the index of the channel the
// value came from and the string value. Process1 returns once all the
// channels are closed.
func Process1(chans []<-chan string, fn func(int, string)) {
    // Setup
    type item struct {
        int    // index of which channel this came from
        string // the actual string item
    }
    merged := make(chan item)
    var wg sync.WaitGroup
    wg.Add(len(chans))
    for i, c := range chans {
        go func(i int, c <-chan string) {
            // Reads and buffers a single item from `c` before
            // we even know if we can write to `merged`.
            //
            // Go doesn't provide a way to do something like:
            //     merged <- (<-c)
            // atomically, where we delay the read from `c`
            // until we can write to `merged`. The read from
            // `c` will always happen first (blocking as
            // required) and then we block on `merged` (with
            // either the above or the below syntax making
            // no difference).
            for s := range c {
                merged <- item{i, s}
            }
            // If/when this input channel is closed we just stop
            // writing to the merged channel and via the WaitGroup
            // let it be known there is one fewer channel active.
            wg.Done()
        }(i, c)
    }
    // One extra goroutine to watch for all the merging goroutines to
    // be finished and then close the merged channel.
    go func() {
        wg.Wait()
        close(merged)
    }()

    // "select-like" loop
    for i := range merged {
        // Process each value
        fn(i.int, i.string)
    }
}

反射选择：

// Process2 is identical to Process1 except that it uses the reflect
// package to select and read from the input channels which guarantees
// there is only one value "in-flight" (i.e. when `fn` is called only
// a single send on a single channel will have succeeded, the rest will
// be blocked). It is approximately two orders of magnitude slower than
// Process1 (which is still insignificant if their is a significant
// delay between incoming values or if `fn` runs for a significant
// time).
func Process2(chans []<-chan string, fn func(int, string)) {
    // Setup
    cases := make([]reflect.SelectCase, len(chans))
    // `ids` maps the index within cases to the original `chans` index.
    ids := make([]int, len(chans))
    for i, c := range chans {
        cases[i] = reflect.SelectCase{
            Dir:  reflect.SelectRecv,
            Chan: reflect.ValueOf(c),
        }
        ids[i] = i
    }

    // Select loop
    for len(cases) > 0 {
        // A difference here from the merging goroutines is
        // that `v` is the only value "in-flight" that any of
        // the workers have sent. All other workers are blocked
        // trying to send the single value they have calculated
        // where-as the goroutine version reads/buffers a single
        // extra value from each worker.
        i, v, ok := reflect.Select(cases)
        if !ok {
            // Channel cases[i] has been closed, remove it
            // from our slice of cases and update our ids
            // mapping as well.
            cases = append(cases[:i], cases[i+1:]...)
            ids = append(ids[:i], ids[i+1:]...)
            continue
        }

        // Process each value
        fn(ids[i], v.String())
    }
}

[围棋操场上的完整代码。]

于 2015-09-03T16:52:33.047 回答
4

可能更简单的选择：

与其拥有一组通道，不如只将一个通道作为参数传递给在单独的 goroutine 上运行的函数，然后在消费者 goroutine 中监听通道？

这允许您在侦听器中仅选择一个通道，进行简单的选择，并避免创建新的 goroutine 来聚合来自多个通道的消息？

于 2020-07-31T07:03:49.970 回答
4

我们实际上对这个主题进行了一些研究，并找到了最佳解决方案。我们使用reflect.Select了一段时间，它是解决问题的好方法。它比每个通道的 goroutine 轻得多，并且操作简单。但不幸的是，它并不真正支持大量的频道，这就是我们的情况，所以我们发现了一些有趣的东西并写了一篇关于它的博客文章：https ://cyolo.io/blog/how-we-enabled-dynamic-channel - 规模化选择/

我将总结那里写的内容：我们静态地创建了一批 select..case 语句，每个结果的指数为 32 的两个幂的每个结果，以及一个路由到不同案例并通过聚合通道聚合结果的函数.

这种批次的一个例子：

func select4(ctx context.Context, chanz []chan interface{}, res chan *r, r *r, i int) {
    select {
    case r.v, r.ok = <-chanz[0]:
        r.i = i + 0
        res <- r
    case r.v, r.ok = <-chanz[1]:
        r.i = i + 1
        res <- r
    case r.v, r.ok = <-chanz[2]:
        r.i = i + 2
        res <- r
    case r.v, r.ok = <-chanz[3]:
        r.i = i + 3
        res <- r
    case <-ctx.Done():
        break
    }
}

以及使用这些select..case批次从任意数量的通道聚合第一个结果的逻辑：

    for i < len(channels) {
        l = len(channels) - i
        switch {
        case l > 31 && maxBatchSize >= 32:
            go select32(ctx, channels[i:i+32], agg, rPool.Get().(*r), i)
            i += 32
        case l > 15 && maxBatchSize >= 16:
            go select16(ctx, channels[i:i+16], agg, rPool.Get().(*r), i)
            i += 16
        case l > 7 && maxBatchSize >= 8:
            go select8(ctx, channels[i:i+8], agg, rPool.Get().(*r), i)
            i += 8
        case l > 3 && maxBatchSize >= 4:
            go select4(ctx, channels[i:i+4], agg, rPool.Get().(*r), i)
            i += 4
        case l > 1 && maxBatchSize >= 2:
            go select2(ctx, channels[i:i+2], agg, rPool.Get().(*r), i)
            i += 2
        case l > 0:
            go select1(ctx, channels[i], agg, rPool.Get().(*r), i)
            i += 1
        }
    }
于 2021-02-09T18:52:26.573 回答
-1

假设有人正在发送事件，为什么这种方法不起作用？

func main() {
    numChans := 2
    var chans = []chan string{}

    for i := 0; i < numChans; i++ {
        tmp := make(chan string)
        chans = append(chans, tmp)
    }

    for true {
        for i, c := range chans {
            select {
            case x = <-c:
                fmt.Printf("received %d \n", i)
                go DoShit(x, i)
            default: continue
            }
        }
    }
}
于 2018-03-14T18:24:17.377 回答