Go语言实现：单生产者多消费者模式（Fan-Out）

碧海醫心

发布时间：2025-10-15 11:43:01

496人浏览过

来源于php中文网

原创

go语言实现：单生产者多消费者模式（fan-out）

本文介绍了如何在Go语言中实现单生产者多消费者模式，也称为 Fan-Out 模式。该模式将单个输入通道的数据复制到多个输出通道，允许不同的消费者并行处理相同的数据。文章提供了两种实现方式：一种使用带缓冲的通道，另一种使用无缓冲的通道，并讨论了缓冲大小对消费者滞后的影响以及如何正确关闭输出通道。

在并发编程中，单生产者多消费者（Fan-Out）模式是一种常见的需求。它允许将单个数据源（生产者）产生的数据分发给多个消费者进行并行处理。在Go语言中，利用其强大的goroutine和channel机制，可以轻松实现这种模式。

Fan-Out 模式的核心：数据复制与分发

Fan-Out 模式的关键在于将单个输入通道的数据复制到多个输出通道。每个输出通道都将接收到输入通道的完整数据流，从而允许不同的消费者独立地处理这些数据。

实现 Fan-Out 模式

以下提供两种实现 Fan-Out 模式的 Go 代码示例：一种使用带缓冲的通道，另一种使用无缓冲的通道。

立即学习“go语言免费学习笔记（深入）”；

1. 使用带缓冲的通道

func fanOut(ch <-chan int, size, lag int) []chan int {
    cs := make([]chan int, size)
    for i := range cs {
        // 通道缓冲区大小控制消费者滞后的程度
        cs[i] = make(chan int, lag)
    }
    go func() {
        for i := range ch {
            for _, c := range cs {
                c <- i
            }
        }
        for _, c := range cs {
            // 当输入通道耗尽时，关闭所有输出通道
            close(c)
        }
    }()
    return cs
}

在这个实现中，fanOut 函数接收一个只读通道 ch 作为输入，以及输出通道的数量 size 和缓冲区大小 lag。它创建一个包含 size 个通道的切片 cs，并为每个通道分配一个大小为 lag 的缓冲区。

然后，它启动一个 goroutine，从输入通道 ch 中读取数据，并将每个数据复制到所有的输出通道 cs 中。重要的一点是，当输入通道 ch 被关闭时，这个 goroutine 会关闭所有的输出通道 cs，这对于避免消费者goroutine无限期地阻塞至关重要。

Synthesys

Synthesys是一家领先的AI虚拟媒体平台，用户只需点击几下鼠标就可以制作专业的AI画外音和AI视频

下载

lag 参数控制了消费者可以滞后于生产者多少。如果 lag 设置得太小，可能会导致生产者阻塞，因为输出通道已满。如果 lag 设置得太大，可能会导致消费者处理的数据过时。

2. 使用无缓冲的通道

func fanOutUnbuffered(ch <-chan int, size int) []chan int {
    cs := make([]chan int, size)
    for i := range cs {
        cs[i] = make(chan int)
    }
    go func() {
        for i := range ch {
            for _, c := range cs {
                c <- i
            }
        }
        for _, c := range cs {
            close(c)
        }
    }()
    return cs
}

与带缓冲的通道实现类似，fanOutUnbuffered 函数也接收一个只读通道 ch 作为输入和输出通道的数量 size。不同之处在于，它创建的输出通道是无缓冲的。

使用无缓冲通道意味着生产者必须等待消费者准备好接收数据，才能继续发送下一个数据。这可以确保消费者不会滞后于生产者太多，但也可能导致生产者阻塞。

示例代码

以下是一个完整的示例代码，演示了如何使用 fanOutUnbuffered 函数实现单生产者多消费者模式：

package main

import (
    "fmt"
    "time"
)

func producer(iters int) <-chan int {
    c := make(chan int)
    go func() {
        for i := 0; i < iters; i++ {
            c <- i
            time.Sleep(1 * time.Second)
        }
        close(c)
    }()
    return c
}

func consumer(cin <-chan int) {
    for i := range cin {
        fmt.Println(i)
    }
}

func fanOutUnbuffered(ch <-chan int, size int) []chan int {
    cs := make([]chan int, size)
    for i := range cs {
        cs[i] = make(chan int)
    }
    go func() {
        for i := range ch {
            for _, c := range cs {
                c <- i
            }
        }
        for _, c := range cs {
            close(c)
        }
    }()
    return cs
}

func main() {
    c := producer(10)
    chans := fanOutUnbuffered(c, 3)
    go consumer(chans[0])
    go consumer(chans[1])
    consumer(chans[2])
}

在这个例子中，producer 函数生成一个包含 10 个整数的通道。fanOutUnbuffered 函数将这个通道的数据复制到 3 个输出通道。然后，启动 3 个 goroutine，每个 goroutine 从一个输出通道中读取数据并打印到控制台。