Golang如何用 io.Pipe 实现流式数据传输_Golang io.Pipe 流处理实战-Golang-PHP中文网

Golang如何用 io.Pipe 实现流式数据传输_Golang io.Pipe 流处理实战

P粉602998670

发布： 2025-11-15 22:08:03

原创

436人浏览过

io.Pipe通过返回PipeReader和PipeWriter实现goroutine间同步数据传输，写入阻塞直至被读取，需在独立协程中进行读写操作以避免死锁，常用于流式处理大文件或网络数据，结合gzip可实现边压缩边传输，适用于内存受限场景，使用时需注意正确关闭管道并传递错误信息以确保异常可被及时捕获与处理。

golang如何用 io.pipe 实现流式数据传输_golang io.pipe 流处理实战

在Go语言中，io.Pipe 是实现流式数据传输的常用方式之一。它提供了一个简单的管道机制，允许在一个goroutine中写入数据，在另一个goroutine中读取数据，非常适合处理大文件、网络流、压缩解压等场景下的实时数据传输。

io.Pipe 基本原理

io.Pipe 返回一个 io.PipeReader 和 io.PipeWriter，它们通过内存中的缓冲区连接。写入到 PipeWriter 的数据可以从 PipeReader 中读出。这种结构是同步的：写操作会阻塞直到有读操作来接收数据（反之亦然），因此必须在不同的goroutine中使用读写端，否则会导致死锁。

典型用法是在一个协程中生成或处理数据并写入 PipeWriter，同时在主协程或其他协程中从 PipeReader 读取并消费数据。

简单示例：模拟流式文本传输

下面是一个基础例子，展示如何使用 io.Pipe 实现字符串的流式传输：

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

腾讯智影-AI数字人

基于AI数字人能力，实现7*24小时AI数字人直播带货，低成本实现直播业务快速增增，全天智能在线直播

查看详情

package main

import (
    "io"
    "log"
    "os"
)

func main() {
    // 创建管道
    r, w := io.Pipe()

    // 单独启动一个goroutine写入数据
    go func() {
        defer w.Close()
        messages := []string{"Hello", "World", "Streaming", "With", "Pipe"}
        for _, msg := range messages {
            _, err := w.Write([]byte(msg + "\n"))
            if err != nil {
                log.Printf("写入错误: %v", err)
                return
            }
        }
    }()

    // 主协程读取数据并输出到标准输出
    defer r.Close()
    _, err := io.Copy(os.Stdout, r)
    if err != nil {
        log.Fatalf("读取错误: %v", err)
    }
}

登录后复制

这个例子中，写端在子协程中逐条发送消息，读端通过 io.Copy 将内容输出到终端。由于管道是线程安全的，且自动处理同步，代码简洁高效。

实战应用：压缩文件流式传输

更实用的场景是结合 gzip 对大文件进行边压缩边传输，避免一次性加载整个文件到内存。

package main

import (
    "compress/gzip"
    "io"
    "log"
    "os"
)

func compressFile(filename string, writer *io.PipeWriter) {
    defer writer.Close()
    file, err := os.Open(filename)
    if err != nil {
        log.Printf("无法打开文件: %v", err)
        return
    }
    defer file.Close()

    gzipWriter := gzip.NewWriter(writer)
    defer gzipWriter.Close()

    _, err = io.Copy(gzipWriter, file)
    if err != nil {
        log.Printf("压缩过程中出错: %v", err)
    }
}

func main() {
    r, w := io.Pipe()

    // 启动压缩协程
    go compressFile("large_input.txt", w)

    // 读取压缩后的数据并保存到新文件
    outFile, err := os.Create("output.gz")
    if err != nil {
        log.Fatal(err)
    }
    defer outFile.Close()

    _, err = io.Copy(outFile, r)
    if err != nil {
        log.Fatalf("写入压缩文件失败: %v", err)
    }

    r.Close()
    log.Println("压缩完成")
}

登录后复制

这里我们把一个大文本文件通过管道流式压缩成 gzip 文件。整个过程不会将全部内容加载进内存，适合处理GB级文件。