如何使用Golang io.Reader和io.Writer组合实现数据管道_流式处理示例

Go 的 io.Reader 和 io.Writer 是流式处理核心接口，通过 io.Pipe、io.TeeReader、io.MultiReader 和 io.Copy 等组合可构建灵活低内存管道：Pipe 实现 goroutine 间双向流，TeeReader 边读边写（如计算哈希），MultiReader 顺序合并多个 Reader，自定义 Reader/Writer 可扩展能力（如进度回调）。

Go 语言的 io.Reader 和 io.Writer 是流式处理的核心接口，它们不关心数据来源或去向，只专注“读”和“写”的契约。利用它们的组合能力（如 io.Pipe、io.MultiReader、io.TeeReader、io.Copy 等），可以构建灵活、低内存占用的数据管道。

用 io.Pipe 构建双向流管道

io.Pipe() 返回一对连接的 io.Reader 和 io.Writer：往 writer 写入的数据可立即被 reader 读出，适合在 goroutine 间传递流式数据，避免内存堆积。

示例：将大文件逐块压缩并写入网络，不全量加载到内存：

pr, pw := io.Pipe()
defer pr.Close()
// 启动压缩 goroutine：从 pr 读，压缩后写入响应
go func() {
defer pw.Close()
gz := gzip.NewWriter(pw)
_, err := io.Copy(gz, pr) // pr 流入 → gz 压缩 → pw 流出
gz.Close()
if err != nil {
log.Printf("compress error: %v", err)
}
}()
// 主流程：打开文件，拷贝到 pw（即流入管道）
file, _ := os.Open("huge.log")
defer file.Close()
io.Copy(pw, file) // file → pw → pr → gz → ...

用 io.TeeReader 实现“边读边记录”管道

io.TeeReader(r, w) 在读取 r 的同时，把读到的数据**原样写入** w（如日志文件、hash 计算器、监控计数器）。它不缓冲，不改变原始流行为，适合审计、校验、埋点等场景。

示例：上传文件时实时计算 SHA256 并保存原始内容：

file, _ := os.Open("upload.bin")
hasher := sha256.New()
tee := io.TeeReader(file, hasher) // 每次 Read 都同步写入 hasher
// 将 tee 流复制到目标存储（如磁盘或对象存储）
dst, _ := os.Create("saved.bin")
n, err := io.Copy(dst, tee)
fmt.Printf("copied %d bytes, hash: %x\n", n, hasher.Sum(nil))

用 io.MultiReader 组合多个 Reader 成单一流

io.MultiReader(rs...io.Reader) 把多个 reader 串成一个逻辑 reader，按顺序读完第一个再读第二个……适合拼接 header + body、多段配置、分片合并等。

示例：构造 HTTP 响应体（JSON header + CSV body）：

header := strings.NewReader(`{"count":3,"format":"csv"}\n`)
body := strings.NewReader("name,age\nAlice,30\nBob,25\n")
combined := io.MultiReader(header, body)
http.ServeContent(w, r, "data.json", time.Now(), combined)

用 io.Copy + 自定义 Reader/Writer 扩展管道能力

只要实现 Read([]byte) (int, error) 或 Write([]byte) (int, error)，就能无缝接入标准管道。例如实现带进度回调的 Writer：

type ProgressWriter struct {
    w       io.Writer
    total   int64
    written int64
    onWrite func(int64, int64) // callback: current, total
}
func (pw *ProgressWriter) Write(p []byte) (n int, err error) {
n, err = pw.w.Write(p)
pw.written += int64(n)
pw.onWrite(pw.written, pw.total)
return
}
// 使用
pw := &ProgressWriter{w: os.Stdout, total: fileSize}
io.Copy(pw, src) // 实时打印进度

# js # json # go # golang # csv # 内存占用 # Error # int # 接口 # 堆 # copy # http # 多个 # 流式 # 自定义 # 回调 # 流管 # 第一个 # 就能 # 第二个 # 不全 # 它不

返回目录在线咨询

上一篇：常见的服务器主机故障有哪些，如何快速排查和解决？
下一篇：Bootstrap轮播图学习使用

您的项目需求