当所有通道都关闭时，跳出一个 select 语句

Question

我有两个独立生成数据的 goroutine，每个将数据发送到一个通道。在我的主 goroutine 中，我想在这些输出进入时使用它们中的每一个，但不关心它们进入的顺序。每个通道在耗尽其输出时会自行关闭。虽然 select 语句是像这样独立使用输入的最佳语法，但我还没有看到一种简洁的方法来循环每个通道，直到两个通道都关闭。

for {
    select {
    case p, ok := <-mins:
        if ok {
            fmt.Println("Min:", p) //consume output
        }
    case p, ok := <-maxs:
        if ok {
            fmt.Println("Max:", p) //consume output
        }
    //default: //can't guarantee this won't happen while channels are open
    //    break //ideally I would leave the infinite loop
                //only when both channels are done
    }
}

我能想到的最好的方法是以下（只是草图，可能有编译错误）：

for {
    minDone, maxDone := false, false
    select {
    case p, ok := <-mins:
        if ok {
            fmt.Println("Min:", p) //consume output
        } else {
            minDone = true
        }
    case p, ok := <-maxs:
        if ok {
            fmt.Println("Max:", p) //consume output
        } else {
            maxDone = true
        }
    }
    if (minDone && maxDone) {break}
}

但是，如果您使用两个或三个以上的频道，这似乎会变得站不住脚。我知道的唯一另一种方法是在 switch 语句中使用 timout case，它要么小到足以冒提前退出的风险，要么在最终循环中注入过多的停机时间。有没有更好的方法来测试 select 语句中的通道？

Answer 1

当我遇到这样的需求时，我采取了以下方法：

var wg sync.WaitGroup
wg.Add(2)

go func() {
  defer wg.Done()
  for p := range mins {
    fmt.Println("Min:", p) 
  }
}()

go func() {
  defer wg.Done()
  for p := range maxs {
    fmt.Println("Max:", p) 
  }
}()

wg.Wait()

我知道这不是单一的 for select 循环，但在这种情况下，我觉得如果没有“if”条件，这更具可读性。

Answer 2

您的示例解决方案效果不佳。一旦其中一个关闭，它总是可以立即进行通信。这意味着你的 goroutine 永远不会屈服，其他通道可能永远不会准备好。您将有效地进入一个无限循环。我在这里贴了一个例子来说明效果：http://play.golang.org/p/rOjdvnji49

那么，我将如何解决这个问题？ nil 通道永远不会为通信做好准备。因此，每次遇到关闭的频道时，您都可以取消该频道，以确保它不再被选中。此处可运行示例：http://play.golang.org/p/8lkV_Hffyj

for {
    select {
    case x, ok := <-ch:
        fmt.Println("ch1", x, ok)
        if !ok {
            ch = nil
        }
    case x, ok := <-ch2:
        fmt.Println("ch2", x, ok)
        if !ok {
            ch2 = nil
        }
    }

    if ch == nil && ch2 == nil {
        break
    }
}

至于怕它变得笨拙，我认为不会。很少有频道同时进入太多地方。这种情况很少出现，以至于我的第一个建议就是处理它。将 10 个通道与 nil 进行比较的长 if 语句并不是尝试在 select 中处理 10 个通道时最糟糕的部分。

Answer 3

我写了一个包，它提供了一个功能来解决这个问题（以及其他几个）：

https://github.com/eapache/channels

https://godoc.org/github.com/eapache/channels

查看Multiplex 函数。它使用反射来缩放到任意数量的输入通道。

Answer 4

为什么不使用 goroutine？随着您的频道逐渐关闭，整个事情变成了一个简单的范围循环。

func foo(c chan whatever, prefix s) {
        for v := range c {
                fmt.Println(prefix, v)
        }
}

// ...

go foo(mins, "Min:")
go foo(maxs, "Max:")

Answer 5

关闭在某些情况下很好，但不是全部。我不会在这里使用它。相反，我只会使用完成的频道：

for n := 2; n > 0; {
    select {
    case p := <-mins:
        fmt.Println("Min:", p)  //consume output
    case p := <-maxs:
        fmt.Println("Max:", p)  //consume output
    case <-done:
        n--
    }
}

操场上的完整工作示例：http://play.golang.org/p/Cqd3lg435y