在之前的两篇文章中,我们分别介绍了 Fanout 和 Fanin。通常情况下,我们将它们一起使用,因为我们有一个数据流,我们希望单独对项目进行操作,并且可以使用并发安全地执行此操作。因此,我们扇出到多个工作线程,然后扇回到单个流。
例如,假设您有一个很大的日志文件。您可以将文件分成多个块,允许每个工作人员同时操作文件的不同部分,然后合并结果。
如果您遵循前两篇文章,这种模式是显而易见的。如果您不确定,请参阅上面的链接。
// produce is simulating our single input as a channel func produce() chan int { ch := make(chan int) go func() { for i := 0; i < 10; i++ { ch <- rand.Intn(50) } fmt.Printf("producer done\n") close(ch) // this is important!!! }() return ch } func worker(id int, jobs chan int, out chan OddEven, wg *sync.WaitGroup) { for value := range jobs { odd := "even" if (value & 1) == 1 { odd = "odd" } out <- OddEven{ Number: value, OddEven: odd, } } close(out) // remember this wg.Done() } // OddEven struct will be the result of the work done by each fanout thread // and be the fanin data type OddEven struct { Number int OddEven string } func fanin(inputs []chan OddEven) chan OddEven { output := make(chan OddEven) var wg sync.WaitGroup for i, input := range inputs { wg.Add(1) // explicit params to capture loop vars go func(id int, input chan OddEven, output chan OddEven, wg *sync.WaitGroup) { for value := range input { output <- value } fmt.Printf("done merging source %d\n", id) wg.Done() }(i, input, output, &wg) } go func() { wg.Wait() close(output) // this is important!!! }() return output } func main() { // simulate the input data stream inputCh := produce() numWorkers := 3 // fan-out to send data items to workers as individual jobs var wg sync.WaitGroup workerResults := make([]chan OddEven, numWorkers) for i := 0; i < numWorkers; i++ { wg.Add(1) workerResults[i] = make(chan OddEven) go worker(i, inputCh, workerResults[i], &wg) } go func() { wg.Wait() }() // fan-in the results results := fanin(workerResults) done := make(chan bool) go func() { for value := range results { fmt.Printf("got %d is %s\n", value.Number, value.OddEven) } close(done) }() <-done fmt.Println("done") }
有一个 Produce() 函数可以创建模拟的数字输入流。
有一个工作函数在输入通道上运行,直到没有更多数据为止。它对每个值“处理”输入数据(确定该值是奇数还是偶数),然后将结果结构发送到输出通道。
请注意,当每个工作人员完成后,它会关闭其结果通道。这对于防止死锁是必要的,因为否则 fanin 操作将休眠等待 chan 上的更多数据。
主线程从 Produce 获取输入流,然后启动许多工作线程,为每个工作线程提供自己的通道,以便发送结果。
这些结果通道随后被发送到 fanin 操作。为了扇入,我们创建一个通道来接收输出,然后为每个工作通道启动一个 goroutine。每个 goroutine 只是在通道上进行迭代,直到没有更多数据为止,然后终止。 请记住,我们关闭了工作线程中的结果通道,这就是允许 for 循环终止的原因
请注意,我们使用 WaitGroup 进行 fanin 过程。这让我们知道所有结果通道的所有结果何时已组合到输出通道中。当发生这种情况时,我们关闭输出通道,以便任何消耗输出的下游线程都可以终止。
有了输出通道中的所有数据,主线程就可以继续显示结果了。请注意,我们使用布尔通道来防止主线程在一切完成之前终止;否则,它将终止进程。
请注意,还有另一种方法可以使用 select 语句进行扇入。这里使用的技术更干净一些,因为我们可以增加或减少工人的数量。
另请注意,我们还没有解决有关 SIGTERM 或 SIGINT 等提前终止的问题。这增加了一点复杂性。
你会如何实现这个? 扇出/扇入模式还有其他实现。请在下面留下您的评论和想法?
谢谢!
这篇文章以及本系列所有文章的代码可以在这里找到
以上是Go 中的扇出-扇南模式的详细内容。更多信息请关注PHP中文网其他相关文章!