关于go中的同步channel问题 - Restoring sequencing

Question

在学习 golang - channel 时遇到了一个问题，使用fan-in函数可以将多个 channel 合并到一个 channel 中，但是合并后的channel取值不一定是有序的（同一批次的fan-in中，有的 channel fan-in 了多次）；
下边超链接的教程可以在多个 channelfan-in 时保证同步，但修改了一下源码后便不同步了，详情看代码～

相关连接：

• golang-talk-ppt
• youtube - Go Concurrency Patterns

源码

import (
    "fmt"
)

func main() {
    c := fanIn(boring("Joe"), boring("Ann"))
    for i := 0; i < 5; i++ {
        msg1 := <-c;fmt.Println(msg1.str)
        msg2 := <-c;fmt.Println(msg2.str)
        msg1.wait <- true
        msg2.wait <- true
    }
    fmt.Println("You're both boring; I'm leaving.")
}

type Message struct {
    str  string
    wait chan bool
}

func fanIn(input1, input2 <-chan Message) <-chan Message {
    c := make(chan Message)
    go func() {for {c <- <-input1}}()
    go func() {for {c <- <-input2}}()
    return c
}


func boring(msg string) <-chan Message {
    var waitForIt = make(chan bool)
    c := make(chan Message)
    go func() {
        for i := 0; ; i++ {
            //time.Sleep(time.Duration(rand.Intn(2e2)) * time.Millisecond)
            c <- Message{fmt.Sprintf("%s: %d", msg, i), waitForIt}
            //time.Sleep(time.Duration(rand.Intn(2e2)) * time.Millisecond)
            <-waitForIt
        }
    }()
    return c
}
// 正常输出同步消息
Joe: 0
Ann: 0
Ann: 1
Joe: 1
Joe: 2
Ann: 2
Ann: 3
Joe: 3
Joe: 4
Ann: 4
You're both boring; I'm leaving.

但做一个小小的改动，输出就不同步了：

// main函数中的循环修改，每次只取一个
for i := 0; i < 10; i++ {
    msg1 := <-c;fmt.Println(msg1.str)
    msg1.wait <- true
}
// 输出 不是同步的
Ann: 0
Ann: 1
Ann: 2
Ann: 3
Ann: 4
Ann: 5
Ann: 6
Joe: 0
Ann: 7
Ann: 8
You're both boring; I'm leaving.

为什么不是同步的呢？在我看来for循环中<-chan两次与一次没有什么不同啊~ ಠ_ಠ
请各位大大指点迷津～

Answer 1

知道原因了～，首先代码中使用的是Unbuffered Channel，每次put操作都需要对应一个take操作。
当使用以下代码时取一个解锁一个，会导致同一个协程不断的抢占锁，这样每次取到可能是同一个协程的。

for i := 0; i < 10; i++ {
    msg1 := <-c;fmt.Println(msg1.str)
    msg1.wait <- true
}

所以核心思想是：同一批次中的消息一起取出，然后在一起解锁。

Answer 2

输出乱序是因为 time.sleep 的时长是随机的，如果某个协程每次随机的时间都比较短则会连续输出。

这段程序时机上有个比较巧妙的设计，在首次执行的时候，第一个先 time.sleep 完成的的协程则顺序将会优先:

for i := 0; i < 5; i++ {
    msg1 := <-c
    // ...
}

第一个输入通道 c 的协程阻塞, 然后等待第二个协程完成:

// ...
msg2 := <-c
// ...

而一旦解锁则 msg1 则 msg1 就会立即占据通道 c:

c <- Message{fmt.Sprintf("%s: %d", msg, i), waitForIt}
time.Sleep(time.Duration(rand.Intn(2e3)) * time.Millisecond)
<-waitForIt // 解锁后立即循环阻塞通道 c

即使 msg2 在第二次执行的过程中 time.sleep 消耗的时间比 msg1 短但是 msg1 已经占据通道 c, 则 msg2 进入阻塞.