Go语言并发实践：Goroutine间的高效通信与模式

package main

import "fmt"
import "time"

func producer1(ch chan int) {
    time.Sleep(100 * time.Millisecond) // 模拟生产时间
    ch <- 10
}

func producer2(ch chan int) {
    time.Sleep(50 * time.Millisecond) // 模拟生产时间
    ch <- 20
}

func main() {
    ch1 := make(chan int)
    ch2 := make(chan int)

    go producer1(ch1)
    go producer2(ch2)

    // Goroutine将顺序接收来自ch1和ch2的数据
    // 它会先阻塞直到ch1有数据，然后阻塞直到ch2有数据
    val1 := <-ch1
    val2 := <-ch2
    fmt.Printf("顺序接收：来自ch1的值 %d，来自ch2的值 %d\n", val1, val2)
}

package main

import (
    "fmt"
    "time"
)

func routineA(out chan string) {
    for i := 0; i < 3; i++ {
        time.Sleep(time.Duration(i+1) * 100 * time.Millisecond)
        out <- fmt.Sprintf("来自RoutineA的消息 %d", i+1)
    }
    close(out) // 发送完毕后关闭Channel
}

func routineB(out chan string) {
    for i := 0; i < 2; i++ {
        time.Sleep(time.Duration(i+1) * 150 * time.Millisecond)
        out <- fmt.Sprintf("来自RoutineB的消息 %d", i+1)
    }
    close(out) // 发送完毕后关闭Channel
}

func main() {
    chA := make(chan string)
    chB := make(chan string)

    go routineA(chA)
    go routineB(chB)

    // 使用select语句从多个Channel接收数据
    for {
        select {
        case msgA, ok := <-chA:
            if !ok { // Channel已关闭且无更多数据
                chA = nil // 将chA设为nil，避免再次尝试从已关闭的Channel读取
                fmt.Println("RoutineA已完成发送。")
                break
            }
            fmt.Println("接收到：", msgA)
        case msgB, ok := <-chB:
            if !ok { // Channel已关闭且无更多数据
                chB = nil // 将chB设为nil
                fmt.Println("RoutineB已完成发送。")
                break
            }
            fmt.Println("接收到：", msgB)
        default:
            // 可选：如果所有case都未就绪，则执行default分支。
            // 如果没有default，select会阻塞直到某个case就绪。
            // fmt.Println("当前无消息，等待中...")
            // time.Sleep(50 * time.Millisecond)
        }

        // 当所有监听的Channel都变为nil时，表示所有生产者都已完成，退出循环
        if chA == nil && chB == nil {
            break
        }
    }

    fmt.Println("所有消息处理完毕。")
}

package main

import (
    "fmt"
    "sync"
    "time"
)

func worker(id int, commands chan string, wg *sync.WaitGroup) {
    defer wg.Done()
    for cmd := range commands { // 从共享Channel接收命令
        fmt.Printf("Worker %d 正在处理命令: %s\n", id, cmd)
        time.Sleep(100 * time.Millisecond) // 模拟处理时间
    }
    fmt.Printf("Worker %d 退出。\n", id)
}

func main() {
    commands := make(chan string)
    var wg sync.WaitGroup

    // 启动3个worker Goroutine，它们都从同一个commands Channel接收数据
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go worker(i, commands, &wg)
    }

    // 多个Goroutine向同一个commands Channel发送数据
    go func() {
        for i := 0; i < 5; i++ {
            commands <- fmt.Sprintf("任务A-%d", i+1)
            time.Sleep(50 * time.Millisecond)
        }
    }()

    go func() {
        for i := 0; i < 5; i++ {
            commands <- fmt.Sprintf("任务B-%d", i+1)
            time.Sleep(70 * time.Millisecond)
        }
        close(commands) // 所有任务发送完毕后关闭Channel
    }()

    wg.Wait() // 等待所有worker完成
    fmt.Println("所有任务已处理完毕。")
}

package main

import (
    "fmt"
    "time"
)

// 定义消息结构体，包含命令和用于回复的Channel
type Command struct {
    Cmd   string      // 命令内容
    Reply chan<- int  // 用于发送回复的Channel
}

func processorRoutine(in chan Command) {
    for cmd := range in {
        fmt.Printf("处理器接收到命令: %s\n", cmd.Cmd)
        // 模拟处理时间
        time.Sleep(100 * time.Millisecond)
        // 根据命令处理结果，发送状态码回给请求者
        if cmd.Cmd == "doSomething" {
            cmd.Reply <- 200 // 成功
        } else {
            cmd.Reply <- 500 // 失败
        }
    }
    fmt.Println("处理器例程退出。")
}

func requesterRoutine(id int, out chan<- Command) {
    // 创建一个临时的回复Channel，只接收int类型
    replyCh := make(chan int)

    // 构造命令，并将回复Channel作为消息的一部分发送
    command := Command{Cmd: fmt.Sprintf("doSomething-%d", id), Reply: replyCh}
    out <- command // 发送请求

    // 等待并接收处理器的回复
    status := <-replyCh
    fmt.Printf("请求者 %d 收到回复状态: %d\n", id, status)
    close(replyCh) // 关闭回复Channel，表明不再需要它
}

func main() {
    commandChannel := make(chan Command) // 主命令Channel

    go processorRoutine(commandChannel) // 启动处理器

    // 启动多个请求者
    for i := 1; i <= 3; i++ {
        go requesterRoutine(i, commandChannel)
    }

    // 等待一段时间，确保所有Goroutine有机会执行
    time.Sleep(1 * time.Second)
    close(commandChannel) // 关闭主命令Channel，通知处理器退出
    time.Sleep(100 * time.Millisecond) // 确保处理器有时间退出
    fmt.Println("主程序退出。")
}