在之前的 2 篇文章中，我们分别介绍了 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 



<p>有一个 produce() 函数可以创建模拟的数字输入流。 </p>

<p>有一个工作函数在输入通道上运行，直到没有更多数据为止。它对每个值“处理”输入数据（确定该值是奇数还是偶数），然后将结果结构发送到输出通道。 </p>

<p>请注意，当每个工作人员完成后，它会关闭其结果通道。这对于防止死锁是必要的，因为否则 fanin 操作将休眠等待 chan 上的更多数据。</p>

<p>主线程从 produce 获取输入流，然后启动许多工作线程，为每个工作线程提供自己的通道，以便发送结果。</p>

<p>这些结果通道随后被发送到 fanin 操作。为了扇入，我们创建一个通道来接收输出，然后为每个工作通道启动一个 goroutine。每个 goroutine 只是在通道上进行迭代，直到没有更多数据为止，然后终止。 <em>请记住，我们关闭了工作线程中的结果通道，这就是允许 for 循环终止的原因</em></p>

<p>请注意，我们使用 waitgroup 进行 fanin 过程。这让我们知道所有结果通道的所有结果何时已组合到输出通道中。当发生这种情况时，我们关闭输出通道，以便任何消耗输出的下游线程都可以终止。</p>

<p>输出通道中的所有数据后，主线程可以继续并显示结果。请注意，我们使用布尔通道来防止主线程在一切完成之前终止；否则，它将终止进程。</p>

<p>请注意，还有另一种方法可以使用 select 语句进行扇入。这里使用的技术更干净一些，因为我们可以增加或减少工人的数量。</p>

<p>另请注意，我们尚未解决有关 sigterm 或 sigint 等提前终止的任何问题。这增加了一点复杂性。</p>

<p>你会如何实现这个？ <em>扇出/扇入模式还有其他实现</em>。请在下面留下您的评论和想法吗？</p>

<p>谢谢！</p>

<p><em>这篇文章以及本系列所有文章的代码可以在这里找到</em></p>