Go语言中并发协程间的高效通信与多通道数据处理

package main

import (
    "fmt"
    "time"
)

// Routine1 模拟从两个通道接收数据
func Routine1(command12 <-chan int, command13 <-chan int) {
    fmt.Println("Routine1: 准备接收数据...")
    // 顺序接收：先从command12接收，再从command13接收
    // 如果某个通道没有数据，将在此处阻塞
    cmd1 := <-command12
    fmt.Printf("Routine1: 从 command12 接收到 %d\n", cmd1)

    cmd2 := <-command13
    fmt.Printf("Routine1: 从 command13 接收到 %d\n", cmd2)

    fmt.Println("Routine1: 成功接收并处理两份数据。")
}

// Routine2 模拟发送数据到 command12
func Routine2(command12 chan<- int) {
    time.Sleep(1 * time.Second) // 模拟一些工作
    data := 100
    command12 <- data
    fmt.Printf("Routine2: 发送 %d 到 command12\n", data)
}

// Routine3 模拟发送数据到 command13
func Routine3(command13 chan<- int) {
    time.Sleep(2 * time.Second) // 模拟一些工作
    data := 200
    command13 <- data
    fmt.Printf("Routine3: 发送 %d 到 command13\n", data)
}

func main() {
    command12 := make(chan int)
    command13 := make(chan int)

    go Routine1(command12, command13)
    go Routine2(command12)
    go Routine3(command13)

    // 确保主协程不会立即退出
    time.Sleep(3 * time.Second)
    fmt.Println("主协程退出。")
}

package main

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

// Routine1 模拟使用 select 从两个通道接收数据
func Routine1Select(command12 <-chan int, command13 <-chan int) {
    for i := 0; i < 5; i++ { // 循环接收5次
        select {
        case cmd1 := <-command12:
            fmt.Printf("Routine1Select: 从 command12 接收到 %d (来自 Routine2)\n", cmd1)
            // 处理来自Routine2的数据
        case cmd2 := <-command13:
            fmt.Printf("Routine1Select: 从 command13 接收到 %d (来自 Routine3)\n", cmd2)
            // 处理来自Routine3的数据
        case <-time.After(500 * time.Millisecond): // 设置超时
            fmt.Println("Routine1Select: 等待超时，没有新的数据到达。")
        }
    }
    fmt.Println("Routine1Select: 接收循环结束。")
}

// Routine2 模拟间歇性发送数据
func Routine2Send(command12 chan<- int) {
    for i := 0; i < 3; i++ {
        time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond) // 随机延迟
        data := rand.Intn(100)
        command12 <- data
        fmt.Printf("Routine2Send: 发送 %d 到 command12\n", data)
    }
    close(command12) // 发送完毕后关闭通道
}

// Routine3 模拟间歇性发送数据
func Routine3Send(command13 chan<- int) {
    for i := 0; i < 3; i++ {
        time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond) // 随机延迟
        data := rand.Intn(100) + 1000
        command13 <- data
        fmt.Printf("Routine3Send: 发送 %d 到 command13\n", data)
    }
    close(command13) // 发送完毕后关闭通道
}

func main() {
    command12 := make(chan int)
    command13 := make(chan int)

    go Routine1Select(command12, command13)
    go Routine2Send(command12)
    go Routine3Send(command13)

    // 确保主协程不会立即退出
    time.Sleep(5 * time.Second)
    fmt.Println("主协程退出。")
}

// 示例：处理通道关闭
select {
case val, ok := <-ch1:
    if !ok {
        fmt.Println("ch1 已关闭")
        return // 或采取其他关闭处理逻辑
    }
    // 处理 val
case val, ok := <-ch2:
    if !ok {
        fmt.Println("ch2 已关闭")
        return
    }
    // 处理 val
}

package main

import (
    "fmt"
    "time"
)

// Command 定义了包含命令字符串和回复通道的消息结构
type Command struct {
    Cmd   string
    Reply chan int // 用于发送响应的通道
}

// Routine1 作为服务提供者，接收 Command 并发送回复
func Routine1Service(commandChan <-chan Command) {
    fmt.Println("Routine1Service: 准备接收命令...")
    for cmd := range commandChan { // 持续从命令通道接收
        fmt.Printf("Routine1Service: 接收到命令 '%s'\n", cmd.Cmd)
        // 模拟处理命令
        status := 200 // 假设处理成功
        time.Sleep(100 * time.Millisecond) // 模拟处理时间

        // 通过消息中携带的回复通道发送状态码
        cmd.Reply <- status
        fmt.Printf("Routine1Service: 发送状态码 %d 到回复通道\n", status)
    }
    fmt.Println("Routine1Service: 命令通道已关闭，服务停止。")
}

// Routine2 作为客户端，发送命令并等待回复
func Routine2Client(commandChan chan<- Command) {
    fmt.Println("Routine2Client: 准备发送命令...")
    // 创建一个用于接收回复的通道
    replyChan := make(chan int)
    cmd := Command{Cmd: "doSomething", Reply: replyChan}

    // 发送命令
    commandChan <- cmd
    fmt.Println("Routine2Client: 发送命令 'doSomething'...")

    // 等待并接收回复
    status := <-replyChan
    fmt.Printf("Routine2Client: 接收到回复状态码 %d\n", status)
    close(replyChan) // 回复通道使用完毕后关闭
}

func main() {
    commandChan := make(chan Command) // 用于发送 Command 结构体的通道

    go Routine1Service(commandChan)
    go Routine2Client(commandChan)

    // 确保所有协程有时间完成
    time.Sleep(2 * time.Second)
    close(commandChan) // 关闭命令通道，通知 Routine1Service 停止
    time.Sleep(500 * time.Millisecond) // 等待 Routine1Service 退出
    fmt.Println("主协程退出。")
}