通道就绪时从多个通道中选择所有值

Question

我是golang 的新手，遇到了问题。

我有几个channels。

一些有效载荷在不同的时间到达这个通道。

如何在频道准备吐出时从频道中一一获取所有值。

例如我写了这段代码：

主包

import (
    "fmt"
    "time"
    "math/rand"
)

func main() {
    arr1 := []int8{1,2,3,4,5}
    arr2 := []int8{6,7,8,9,10}

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

    go func() {
        for _, val := range arr1 {
            time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
            c1 <- val
        }
    }()
    go func() {
        for _, val := range arr2 {
            time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
            c2 <- val
        }
    }()

    select {
        case res1 := <- c1:
            fmt.Println(res1)
        case res2 := <- c2:
            fmt.Println(res2)
    }

    fmt.Println("Hello, test")
}

但在这种情况下，我只从其中一个渠道获得第一个值。

请给我建议如何解决我的问题。

游戏链接https://play.golang.org/p/FOmkP57YCyR

Answer 1

你必须做几件事。

1) 确保在完成源后关闭频道。 2) 遍历通道直到它被关闭。

例子：

package main

import (
    "fmt"
    "math/rand"
    "time"
)

func main() {
    arr1 := []int8{1, 2, 3, 4, 5}
    arr2 := []int8{6, 7, 8, 9, 10}

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

    go func() {
        for _, val := range arr1 {
            time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
            c1 <- val
        }
        close(c1)
    }()
    go func() {
        for _, val := range arr2 {
            time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
            c2 <- val
        }
        close(c2)
    }()

    _c1 := true
    _c2 := true
    var res1, res2 int8

    for _c1 == true || _c2 == true {
        select {
        case res1, _c1 = <-c1:
            if _c1 == true {
                fmt.Println(res1)
            }
        case res2, _c2 = <-c2:
            if _c2 == true {
                fmt.Println(res2)
            }
        }
    }

    fmt.Println("Hello, test")
}

在执行时，我在屏幕上得到以下输出。

6
1
7
2
3
4
8
5
9
10
Hello, test

Answer 2

您不必使用 2 个频道。只需使用 1 个通道并将来自多个 goroutine 的值存储到其中。 Channel 是轻量级的线程连接器，速度快，可以多次实例化以存储来自多个 goroutine 的值。

您的代码的问题在于它没有循环来遍历来自 goroutines 通道的值。您只需使用 select 打印一次。 select 让其他 goroutine 等到它执行它所拥有的一种可能情况。如果所有情况都可能，则随机选择执行。

你只从你的通道中获得一个值的原因是因为当你的 goroutines 工作时，它们将数组中的值按顺序存储到通道中。当这种情况发生时，你在你的主线程中调用select 语句，并在你的goroutines中从通道中获取值。由于您不遍历通道，因此您只会从通道中获得一个值，即它首先收到的值。在这种情况下，您将数组顺序循环到 goroutine 中的通道中，因此您将获得数组的第一个索引，因为它是首先发送到主线程中的 select 语句的值。您选择的所有案例都可以执行，因此它将随机执行其中一个案例，您将在其中一个数组中获得第一个索引。

要解决此问题，您需要遍历通道以逐一获取存储在其中的值。此外，您还需要同步所有线程以避免死锁情况，当您的主线程不知道何时停止从 goroutines 调用通道时会发生这种情况，因为它们是异步工作的。您的通道在 goroutine 中获取其值并在循环中调用到主线程后立即准备吐出该值。代码如下：

package main

import (
    "fmt"
    "time"
    "math/rand"
    "sync"
)

// writer set numbers from array to channel
func writer(ch chan int, arr []int ,wgwrite *sync.WaitGroup) {
    defer wgwrite.Done()
    for _, val := range arr {
        time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
        ch <- val
    }
}

// reader receive input from writer channels and print them all 
func reader(ch chan int, wgread *sync.WaitGroup) {
  defer wgread.Done()
  for i:= range ch {
    fmt.Println(i)
  }
  fmt.Println("Hello, test")
}

func main() {
    arr1 := []int{1,2,3,4,5}
    arr2 := []int{6,7,8,9,10}
    ch := make(chan int)
    wgwrite := &sync.WaitGroup{}
    wgread  := &sync.WaitGroup{}

    wgwrite.Add(2)
    go writer(ch, arr1, wgwrite)
    go writer(ch, arr2, wgwrite)

    wgread.Add(1)
    go reader(ch, wgread)

    wgwrite.Wait()
    close(ch)
    wgread.Wait()
}

https://play.golang.org/p/32Fgetq_Zu7

希望对你有帮助。

Answer 3

Select 不等待执行例程。要实现这一点，您应该将其包装在 for 语句中。这样select 将一直运行，直到其中一种情况返回并突破for 语句。

for {
   select {
      ...

您还可以使用非阻塞和等待组的缓冲通道。像这样：

arr1 := []int8{1,2,3,4,5}
arr2 := []int8{6,7,8,9,10}

c1 := make(chan int8, len(arr1))
c2 := make(chan int8, len(arr2))

var wg sync.WaitGroup

wg.Add(1) // First wait group
go func() {
    for _, val := range arr1 {
        time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
        c1 <- val
    }
    wg.Done()
}()

wg.Add(1) // Second wait group
go func() {
    for _, val := range arr2 {
        time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
        c2 <- val
    }
    wg.Done()
}()

// executed after wg.Done() is called 2 times since we have 2 wait groups
wg.Wait() 

// We are not writing to channels anymore so we can close them.
close(c1)
close(c2)

for value := range c1 {
    fmt.Println(value)
}

for value := range c2 {
    fmt.Println(value)
}


fmt.Println("Hello, test")