面试题：Go函数式反应型编程中处理并发数据流

Question

在Go函数式反应型编程场景下，假设有多个并发的数据流（以通道表示），每个数据流会源源不断地产生整数。要求使用函数式编程的思想，将这些数据流中的所有整数进行累加，并在所有数据流结束时返回最终结果。请描述实现思路并编写核心代码。

Answer 1

1. 实现思路：
   • 使用 sync.WaitGroup 来等待所有数据流处理完毕。
   • 为每个数据流创建一个独立的 goroutine 来读取数据并进行累加。
   • 利用 select 语句在多个通道间进行非阻塞的读取操作，当所有通道都关闭时，结束累加操作并返回结果。
2. 核心代码：

package main

import (
    "fmt"
    "sync"
)

func sumChannels(channels []<-chan int) int {
    var wg sync.WaitGroup
    total := 0
    for _, ch := range channels {
       wg.Add(1)
       go func(c <-chan int) {
          defer wg.Done()
          for num := range c {
             total += num
          }
       }(ch)
    }
    go func() {
       wg.Wait()
       close(done)
    }()
    for {
       select {
       case <-done:
          return total
       }
    }
}

你可以这样调用这个函数：

func main() {
    ch1 := make(chan int)
    ch2 := make(chan int)
    go func() {
       ch1 <- 1
       ch1 <- 2
       close(ch1)
    }()
    go func() {
       ch2 <- 3
       ch2 <- 4
       close(ch2)
    }()
    result := sumChannels([]<-chan int{ch1, ch2})
    fmt.Println("Total:", result)
}

其中 sumChannels 函数实现了对多个通道数据的累加，main 函数给出了使用示例。这里还使用了一个 done 通道来通知所有数据处理完毕。完整代码中应在包级别定义 done 通道：

var done = make(chan struct{})