多路复用多个 Go Channel 到单个 Channel

本文介绍了如何将多个只读的 Go Channel 中的数据合并到一个单独的只写 Channel 中，并在所有输入 Channel 关闭后关闭输出 Channel。通过使用 sync.WaitGroup 来同步 Goroutine，确保所有输入 Channel 的数据都被处理完毕后才关闭输出 Channel，避免数据丢失，提供了一种高效且安全的 Channel 多路复用方案。

在并发编程中，经常会遇到需要将多个 Channel 的数据汇聚到一个 Channel 的场景，例如，从多个数据源读取数据，然后统一处理。Go 语言的 Channel 提供了强大的并发通信机制，结合 Goroutine 和 select 语句可以实现 Channel 的多路复用。然而，当需要处理的 Channel 数量不确定时，手动编写 select 语句会变得非常繁琐。本文将介绍一种利用 sync.WaitGroup 实现任意数量 Channel 多路复用的方法。

使用 sync.WaitGroup 实现 Channel 多路复用

以下代码展示了如何使用 sync.WaitGroup 将多个 <-chan int (只读 int Channel) 的数据合并到一个 chan<- int (只写 int Channel) 中，并在所有输入 Channel 关闭后关闭输出 Channel。

package main

import (
    "fmt"
    "sync"
)

func combine(inputs []<-chan int, output chan<- int) {
    var group sync.WaitGroup

    // 启动 Goroutine 处理每个输入 Channel
    for i := range inputs {
        group.Add(1)
        go func(input <-chan int) {
            defer group.Done() // 确保 Goroutine 退出时调用 Done()

            for val := range input {
                output <- val // 将输入 Channel 的数据发送到输出 Channel
            }
            // 输入 Channel 关闭后，该 Goroutine 会自动退出
        }(inputs[i])
    }

    // 启动 Goroutine 等待所有输入 Channel 处理完毕
    go func() {
        group.Wait()       // 等待所有 Goroutine 调用 Done()
        close(output) // 关闭输出 Channel
    }()
}

func main() {
    // 创建多个输入 Channel
    numChannels := 3
    inputChannels := make([]<-chan int, numChannels)
    for i := 0; i < numChannels; i++ {
        ch := make(chan int)
        inputChannels[i] = ch
        go func(index int, ch chan int) {
            for j := 0; j < 5; j++ {
                ch <- index*10 + j
            }
            close(ch)
        }(i, ch)
    }

    // 创建输出 Channel
    outputChannel := make(chan int)

    // 多路复用输入 Channel 到输出 Channel
    combine(inputChannels, outputChannel)

    // 从输出 Channel 读取数据并打印
    for val := range outputChannel {
        fmt.Println(val)
    }

    fmt.Println("Done!")
}

代码解释：

喵记多

喵记多 - 自带助理的 AI 笔记

27

查看详情

1. combine 函数：
   • 接收一个 []<-chan int 类型的输入 Channel 切片和一个 chan<- int 类型的输出 Channel。
   • 使用 sync.WaitGroup 来同步 Goroutine 的完成状态。
   • 为每个输入 Channel 启动一个 Goroutine，负责从该 Channel 读取数据并发送到输出 Channel。
   • 每个 Goroutine 在退出前调用 group.Done()，表示该 Channel 的数据已处理完毕。
   • 启动一个单独的 Goroutine，等待所有输入 Channel 的 Goroutine 完成后关闭输出 Channel。
2. main 函数：
   • 创建多个输入 Channel，每个 Channel 模拟一个数据源。
   • 创建输出 Channel，用于接收所有输入 Channel 的数据。
   • 调用 combine 函数将多个输入 Channel 多路复用到输出 Channel。
   • 从输出 Channel 读取数据并打印，直到输出 Channel 关闭。

注意事项：

• 必须在每个处理输入 Channel 的 Goroutine 中使用 defer group.Done() 确保 Goroutine 退出时调用 Done()，避免 group.Wait() 永久阻塞。
• 输出 Channel 必须在所有输入 Channel 处理完毕后才能关闭，否则可能会丢失数据。
• 输入 Channel 必须是只读 Channel (<-chan int)，输出 Channel 必须是只写 Channel (chan<- int)，以保证数据流向的正确性。

总结：

通过使用 sync.WaitGroup，我们可以轻松地将任意数量的 Channel 多路复用到一个 Channel 中，并在所有输入 Channel 关闭后安全地关闭输出 Channel。这种方法避免了手动编写复杂的 select 语句，提高了代码的可维护性和可读性。该方案适用于各种需要将多个数据源的数据汇聚到一起进行处理的场景。

以上就是多路复用多个 Go Channel 到单个 Channel的详细内容，更多请关注php中文网其它相关文章！