Go并发模式：理解通道执行顺序与消息序列化

package main

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

// Message 结构体包含字符串内容和一个用于同步的通道
type Message struct {
    str  string
    wait chan bool // 用于等待客户端信号的通道
}

// boring 函数模拟一个消息生产者
// 它返回一个只读的 Message 通道
func boring(name string) <-chan Message {
    c := make(chan Message)
    // 关键点：每个 boring Goroutine 拥有自己的 wait 通道实例
    // 这个通道是无缓冲的，意味着发送和接收操作会阻塞直到另一端就绪
    waitForIt := make(chan bool) 

    go func() {
        for i := 0; ; i++ {
            c <- Message{fmt.Sprintf("%s: Iteration %d", name, i), waitForIt}
            time.Sleep(time.Duration(rand.Intn(2e2)) * time.Millisecond) // 模拟工作耗时
            <-waitForIt // 生产者在这里等待客户端的信号，收到信号后才能继续
        }
    }()
    return c
}

// fanIn 函数将多个输入通道的消息汇聚到一个输出通道
func fanIn(input1, input2 <-chan Message) <-chan Message {
    c := make(chan Message)
    go func() {
        for {
            select {
            case s := <-input1:
                c <- s
            case s := <-input2:
                c <- s
            }
        }
    }()
    return c
}

func main() {
    fmt.Println("--- 场景一：仅发送一个等待信号 (错误序列) ---")
    joe := boring("Message 1")
    ann := boring("Message 2")
    c := fanIn(joe, ann)

    fmt.Println("期望输出: Message 1: Iteration 0, Message 2: Iteration 0, Message 1: Iteration 1, Message 2: Iteration 1 ...")
    fmt.Println("实际输出 (仅发送一个等待信号):")
    for i := 0; i < 5; i++ {
        msg1 := <-c // 接收第一个消息 (例如，来自 Joe)
        fmt.Printf("%s\n", msg1.str)
        msg2 := <-c // 接收第二个消息 (例如，来自 Ann)
        fmt.Printf("%s\n", msg2.str)

        // 错误场景：只向 msg1 的 wait 通道发送信号
        msg1.wait <- true // 假设 msg1 来自 Joe，Joe 被解锁
        // msg2.wait <- true // Ann 的 Goroutine 仍然阻塞
    }
    time.Sleep(100 * time.Millisecond) // 留出时间观察效果
    fmt.Println("...")
    fmt.Println("--- 场景一结束 ---")
}

--- 场景一：仅发送一个等待信号 (错误序列) ---
期望输出: Message 1: Iteration 0, Message 2: Iteration 0, Message 1: Iteration 1, Message 2: Iteration 1 ...
实际输出 (仅发送一个等待信号):
Message 1: Iteration 0
Message 2: Iteration 0
Message 1: Iteration 1
Message 1: Iteration 2  // 出现重复，Message 1 连续出现
Message 2: Iteration 1
Message 1: Iteration 3
Message 2: Iteration 2
...
--- 场景一结束 ---

func main() {
    // ... (省略场景一代码) ...

    fmt.Println("\n--- 场景二：发送两个等待信号 (正确序列) ---")
    joe := boring("Message 1")
    ann := boring("Message 2")
    c := fanIn(joe, ann)

    fmt.Println("期望输出: Message 1: Iteration 0, Message 2: Iteration 0, Message 1: Iteration 1, Message 2: Iteration 1 ...")
    fmt.Println("实际输出 (发送两个等待信号):")
    for i := 0; i < 5; i++ {
        msg1 := <-c // 接收第一个消息
        fmt.Printf("%s\n", msg1.str)
        msg2 := <-c // 接收第二个消息
        fmt.Printf("%s\n", msg2.str)

        // 正确场景：分别向 msg1 和 msg2 的 wait 通道发送信号
        msg1.wait <- true // 解锁 msg1 所属的生产者
        msg2.wait <- true // 解锁 msg2 所属的生产者
    }
    time.Sleep(100 * time.Millisecond) // 留出时间观察效果
    fmt.Println("--- 场景二结束 ---")
}

--- 场景二：发送两个等待信号 (正确序列) ---
期望输出: Message 1: Iteration 0, Message 2: Iteration 0, Message 1: Iteration 1, Message 2: Iteration 1 ...
实际输出 (发送两个等待信号):
Message 1: Iteration 0
Message 2: Iteration 0
Message 1: Iteration 1
Message 2: Iteration 1
Message 1: Iteration 2
Message 2: Iteration 2
Message 1: Iteration 3
Message 2: Iteration 3
Message 1: Iteration 4
Message 2: Iteration 4
--- 场景二结束 ---