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无限期地阻塞至关重要。

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 从一个输出通道中读取数据并打印到控制台。