当涉及多个通道时，选择如何工作？

Question

我在多个非缓冲通道上使用 select 时发现

select {
case <- chana:
case <- chanb:
}

即使两个通道都有数据，但是在处理这个选择时， case chana 和 case chanb 的呼叫不平衡。

package main

import (
    "fmt"
    _ "net/http/pprof"
    "sync"
    "time"
)

func main() {
    chana := make(chan int)
    chanb := make(chan int)

    go func() {
        for i := 0; i < 1000; i++ {
            chana <- 100 * i
        }
    }()

    go func() {
        for i := 0; i < 1000; i++ {
            chanb <- i
        }
    }()

    time.Sleep(time.Microsecond * 300)

    acount := 0
    bcount := 0
    wg := sync.WaitGroup{}
    wg.Add(1)
    go func() {
        for {
            select {
            case <-chana:
                acount++
            case <-chanb:
                bcount++
            }
            if acount == 1000 || bcount == 1000 {
                fmt.Println("finish one acount, bcount", acount, bcount)
                break
            }
        }
        wg.Done()
    }()

    wg.Wait()
}

运行这个demo，当一个chana,chanb完成读/写时，另一个可能还剩下999-1。

有什么方法可以保证平衡吗？

找到相关主题
golang-channels-select-statement

Answer 1

Go select 声明不偏向于任何（就绪）案例。引用规范：

如果一个或多个通信可以继续，则通过统一伪随机选择选择一个可以继续的通信。否则，如果存在默认情况，则选择该情况。如果没有默认情况，“select”语句会阻塞，直到至少有一个通信可以继续。

如果可以进行多个通信，则随机选择一个。这不是一个完美的随机分布，规范也不能保证，但它是随机的。

您所体验的是 Go Playground 具有 GOMAXPROCS=1 (which you can verify here) 和 goroutine 调度程序没有抢占式的结果。这意味着默认情况下 goroutine 不会并行执行。如果遇到阻塞操作（例如从网络读取，或尝试在阻塞的通道上接收或发送），则将 goroutine 置于停放状态，而另一个准备运行的程序继续运行。

而且由于您的代码中没有阻塞操作，goroutines 可能不会被放置，它可能只有一个“生产者”goroutines 会运行，而另一个可能不会被调度（永远）。

在GOMAXPROCS=4 的本地计算机上运行您的代码，我得到了非常“真实”的结果。运行几次，输出：

finish one acount, bcount 1000 901
finish one acount, bcount 1000 335
finish one acount, bcount 1000 872
finish one acount, bcount 427 1000

如果您需要优先处理单个案例，请查看此答案：Force priority of go select statement

select 的默认行为不保证同等优先级，但平均而言会接近。如果您需要保证同等优先级，那么您不应该使用select，但您可以从 2 个通道执行 2 个非阻塞接收序列，如下所示：

for {
    select {
    case <-chana:
        acount++
    default:
    }
    select {
    case <-chanb:
        bcount++
    default:
    }
    if acount == 1000 || bcount == 1000 {
        fmt.Println("finish one acount, bcount", acount, bcount)
        break
    }
}

如果两个提供值，上述 2 个非阻塞接收将以相同的速度（具有相同的优先级）耗尽 2 个通道，如果一个不提供，则持续接收另一个通道，而不会延迟或阻塞。

对此需要注意的一点是，如果 没有 通道提供任何要接收的值，这将基本上是一个“忙碌”循环，因此会消耗计算能力。为了避免这种情况，我们可能会检测到没有一个通道准备好，然后然后对两个接收使用select 语句，然后将阻塞直到其中一个准备好接收，不浪费任何 CPU 资源：

for {
    received := 0
    select {
    case <-chana:
        acount++
        received++
    default:
    }
    select {
    case <-chanb:
        bcount++
        received++
    default:
    }

    if received == 0 {
        select {
        case <-chana:
            acount++
        case <-chanb:
            bcount++
        }
    }

    if acount == 1000 || bcount == 1000 {
        fmt.Println("finish one acount, bcount", acount, bcount)
        break
    }
}

有关 goroutine 调度的更多详细信息，请参阅以下问题：

Number of threads used by Go runtime

Goroutines 8kb and windows OS thread 1 mb

Why does it not create many threads when many goroutines are blocked in writing file in golang?

Answer 2

正如评论中提到的，如果你想确保平衡，你可以在阅读 goroutine 中完全放弃使用select 并依赖无缓冲通道提供的同步：

go func() {
    for {
        <-chana
        acount++
        <-chanb
        bcount++

        if acount == 1000 || bcount == 1000 {
            fmt.Println("finish one acount, bcount", acount, bcount)
            break
        }
    }
    wg.Done()
}()

Answer 3

已编辑：您也可以从供应方面进行平衡，但@icza 的回答对我来说似乎是比这更好的选择，并且还解释了首先导致这种情况的调度。令人惊讶的是，即使在我的（虚拟）机器上也是片面的。

这里有一些东西可以平衡供应方面的两个例程（不知何故似乎在 Playground 上不起作用）。

package main

import (
    "fmt"
    _ "net/http/pprof"
    "sync"
    "sync/atomic"
    "time"
)

func main() {
    chana := make(chan int)
    chanb := make(chan int)
    var balanceSwitch int32

    go func() {
        for i := 0; i < 1000; i++ {
            for atomic.LoadInt32(&balanceSwitch) != 0 {
                fmt.Println("Holding R1")
                time.Sleep(time.Nanosecond * 1)
            }
            chana <- 100 * i
            fmt.Println("R1: Sent i", i)
            atomic.StoreInt32(&balanceSwitch, 1)

        }
    }()

    go func() {
        for i := 0; i < 1000; i++ {

            for atomic.LoadInt32(&balanceSwitch) != 1 {
                fmt.Println("Holding R2")
                time.Sleep(time.Nanosecond * 1)
            }
            chanb <- i
            fmt.Println("R2: Sent i", i)
            atomic.StoreInt32(&balanceSwitch, 0)

        }
    }()

    time.Sleep(time.Microsecond * 300)

    acount := 0
    bcount := 0
    wg := sync.WaitGroup{}
    wg.Add(1)
    go func() {
        for {
            select {
            case <-chana:
                acount++
            case <-chanb:
                bcount++
            }
            fmt.Println("Acount Bcount", acount, bcount)
            if acount == 1000 || bcount == 1000 {
                fmt.Println("finish one acount, bcount", acount, bcount)
                break
            }
        }
        wg.Done()
    }()

    wg.Wait()
}

通过更改atomic.LoadInt32(&balanceSwitch) != XX 和atomic.StoreInt32(&balanceSwitch, X) 或其他机制，您可以将其映射到任意数量的例程。这可能不是最好的做法，但如果这是一个要求，那么您可能必须考虑这些选项。希望这会有所帮助。