Go 语言的标准库将输入输出抽象为两个极简接口：io.Reader 和 io.Writer。前者只要求实现 Read(p []byte) (n int, err error)，后者只要求实现 Write(p []byte) (n int, err error)。这两个接口看似简单，却支撑起了文件、网络、压缩、加密等几乎所有场景的流式处理。本文围绕如何把多个 Reader 和 Writer 组合成数据管道展开，从接口语义、基础管道、包装器到并发管道，给出可以直接运行的代码示例。

io.Reader 与 io.Writer 的接口语义

理解数据管道的第一步是把这两个接口的约定吃透。io.Reader 的 Read 方法会读取数据到调用方传入的字节切片中，返回实际读取的字节数 n 和可能的错误 err。当数据暂时不可用但流未结束时，Read 可能返回 n=0, err=nil，调用方应该继续读取；当流正常结束时返回 n=0, err=io.EOF；当发生底层故障时返回非 nil 错误。换句话说，io.EOF 并非异常，而是表示“数据读完了”的普通信号。

io.Writer 的约定同样简单：Write 接收一个字节切片，写入底层数据源，返回写入的字节数和错误。如果写入的字节数小于切片长度，就必须返回 io.ErrShortWrite 或具体错误，调用方可以据此判断是否需要重试。两个接口都只面向字节流，不关心数据来源或去向，这种抽象让组合成为可能。

下面先看一个最小实现。我们定义一个 stringReader 类型，它从字符串中按固定步长读取数据，模拟一个数据源。代码中通过取地址方式创建读取器实例，循环读取直到 io.EOF。

package main
import (
    "fmt"
    "io"
)
type stringReader struct {
    data string
    pos  int
}
func (r *stringReader) Read(p []byte) (int, error) {
    if r.pos >= len(r.data) {
        return 0, io.EOF
    }
    n := copy(p, r.data[r.pos:])
    r.pos += n
    return n, nil
}
func main() {
    r := &stringReader{data: "hello world"}
    buf := make([]byte, 5)
    for {
        n, err := r.Read(buf)
        fmt.Printf("读取 %d 字节：%s\n", n, buf[:n])
        if err == io.EOF {
            break
        }
        if err != nil {
            panic(err)
        }
    }
}

这个例子说明：Reader 不需要一次性提供全部数据，调用方可以用很小的缓冲区循环读取，这正是流式处理的核心。

用 io.Copy 和 io.Pipe 连接流

io.Copy 是标准库提供的最常用的管道连接函数，它接受一个 Writer 和一个 Reader，把从 Reader 读取的所有数据复制到 Writer，直到 Reader 返回 io.EOF。它内部会维护一个 32KB 的缓冲区，避免频繁的小块写入，所以即使没有手动管理缓冲区，性能也足够好。

光有 io.Copy 只能完成一对一的流传输，真正的管道需要把一个流的输出变成另一个流的输入。io.Pipe 提供了这种能力：io.Pipe() 返回一个 *PipeReader 和一个 *PipeWriter，它们共享同一个同步的内存缓冲区。写入方调用 pw.Write 会阻塞，直到读取方调用 pr.Read 取走数据；读取方调用 pr.Read 也会阻塞，直到写入方提供数据或关闭写入端。这种同步特性让管道天然适合在两个 goroutine 之间传递数据。

下面的例子用 io.Pipe 连接一个生产者 goroutine 和一个消费者 goroutine。生产者每秒写入一行日志，消费者读取并打印。

package main
import (
    "fmt"
    "io"
    "time"
)
func main() {
    pr, pw := io.Pipe()
    go func() {
        defer pw.Close()
        for i := 0; i < 3; i++ {
            fmt.Fprintf(pw, "line %d\n", i)
            time.Sleep(100 * time.Millisecond)
        }
    }()
    buf := make([]byte, 64)
    for {
        n, err := pr.Read(buf)
        if n > 0 {
            fmt.Print(string(buf[:n]))
        }
        if err == io.EOF {
            break
        }
        if err != nil {
            panic(err)
        }
    }
}

在这个管道中，io.Pipe 保证数据的顺序和完整性。需要注意的是，PipeWriter.Close 会向读取方发送 io.EOF，而如果读取方提前关闭，写入方会收到 io.ErrClosedPipe。这种紧耦合使得管道中的任何一端都不能随意退出。

包装器组合：在流中插入处理逻辑

管道的主要价值是在数据传输过程中插入转换、过滤、统计等逻辑。Go 的接口设计让这类扩展非常自然：你只需要实现一个包装类型，嵌入原始 io.Reader 或 io.Writer，然后在 Read 或 Write 方法的前后添加自己的逻辑，再把调用转发给底层对象。

例如，定义一个 upperWriter，它把每次写入的字节转换成大写后再交给底层的 io.Writer。这样无需修改数据源或目标，就能在管道中加入转换步骤。

package main
import (
    "bytes"
    "fmt"
    "io"
)
type upperWriter struct {
    dst io.Writer
}
func (u *upperWriter) Write(p []byte) (int, error) {
    transformed := bytes.ToUpper(p)
    return u.dst.Write(transformed)
}
func main() {
    var buf bytes.Buffer
    uw := &upperWriter{dst: &buf}
    io.WriteString(uw, "hello, pipe\n")
    fmt.Println(buf.String())
}

运行这段代码，输出结果为 HELLO, PIPE。这里 bytes.ToUpper 会分配新的切片，如果数据量很大可以考虑原地转换以降低内存分配。包装器的思路同样适用于 Reader：你可以包装一个 Reader，在读取后对数据进行解压、解密或统计。

标准库中已经有许多现成的包装器，比如 bufio.Reader 提供缓冲读取、bufio.Writer 提供缓冲写入、gzip.NewReader 解压 gzip 流、base64.NewDecoder 解码 Base64 流。把它们像积木一样组合，就能构建出强大的数据管道。

构建完整的流式处理管道示例

有了接口、连接函数和包装器，我们就可以构建一个完整的管道。假设需求是：读取一个大文件，对每一行做 Base64 编码，然后写入另一个文件。如果直接读取整个文件再处理，内存占用可能达到文件大小的数倍。用流式管道则可以在固定内存下完成。

实现思路如下：用 os.Open 打开源文件得到 io.Reader，用 bufio.NewReader 包装增加缓冲，用 base64.NewEncoder 包装写入端实现编码，最后通过 io.Copy 把读取端和编码写入端连接起来。编码器会在 Close 时写入结尾数据，所以需要显式关闭。

package main
import (
    "bufio"
    "encoding/base64"
    "io"
    "os"
)
func main() {
    src, err := os.Open("source.txt")
    if err != nil {
        panic(err)
    }
    defer src.Close()
    dst, err := os.Create("encoded.txt")
    if err != nil {
        panic(err)
    }
    defer dst.Close()
    reader := bufio.NewReaderSize(src, 64*1024)
    encoder := base64.NewEncoder(base64.StdEncoding, dst)
    defer encoder.Close()
    if _, err := io.Copy(encoder, reader); err != nil {
        panic(err)
    }
}

这段代码的内存使用几乎与文件大小无关。bufio.ReaderSize 设置了 64KB 的读取缓冲，io.Copy 内部使用 32KB 的复制缓冲，base64.Encoder 内部也有自己的小块缓冲。整个链路中数据像水流一样从源文件流入编码器，再流入目标文件。

如果需要在管道中间同时做校验，可以再插入一个 io.TeeReader。它读取数据时会把数据同步写入一个 Writer，这样主干数据继续流向编码器，旁路数据可以用来计算哈希或记录日志。下面的代码在复制的同时计算 SHA256 校验和。

package main
import (
    "crypto/sha256"
    "fmt"
    "io"
    "os"
)
func main() {
    src, _ := os.Open("source.txt")
    defer src.Close()
    hash := sha256.New()
    tee := io.TeeReader(src, hash)
    dst, _ := os.Create("copy.txt")
    defer dst.Close()
    if _, err := io.Copy(dst, tee); err != nil {
        panic(err)
    }
    fmt.Printf("SHA256: %x\n", hash.Sum(nil))
}

io.TeeReader 把读取到的每个字节同时写入 hash，因为 hash.Hash 实现了 io.Writer 接口。这样主干复制到目标文件，旁路完成哈希计算，两条路径互不干扰。

并发管道的错误处理与资源释放

管道一旦涉及多个 goroutine，错误传播和资源释放就会变得复杂。以 io.Pipe 为例，如果生产者在写入过程中遇到错误，不能简单地直接 return，否则读取方会一直阻塞等待数据。正确做法是调用 pw.CloseWithError(err)，把错误传递给读取方；读取方在 Read 时会返回这个错误，从而退出循环。

同样，如果消费者先退出，生产者写入时也会收到 io.ErrClosedPipe，避免 goroutine 泄漏。下面是一个使用 io.Pipe 和 context.Context 实现可取消管道的示例。

package main
import (
    "context"
    "io"
    "os"
    "time"
)
func produce(ctx context.Context, pw *io.PipeWriter) {
    defer pw.Close()
    for {
        select {
        case <-ctx.Done():
            pw.CloseWithError(ctx.Err())
            return
        default:
            if _, err := pw.Write([]byte("data\n")); err != nil {
                return
            }
            time.Sleep(100 * time.Millisecond)
        }
    }
}
func consume(ctx context.Context, pr *io.PipeReader) {
    buf := make([]byte, 64)
    for {
        select {
        case <-ctx.Done():
            pr.CloseWithError(ctx.Err())
            return
        default:
            n, err := pr.Read(buf)
            if n > 0 {
                os.Stdout.Write(buf[:n])
            }
            if err != nil {
                if err != io.EOF {
                    _ = err
                }
                return
            }
        }
    }
}
func main() {
    ctx, cancel := context.WithTimeout(context.Background(), time.Second)
    defer cancel()
    pr, pw := io.Pipe()
    go produce(ctx, pw)
    go consume(ctx, pr)
    <-ctx.Done()
}

这里的 produce 和 consume 通过 context.Context 协调退出。当超时触发时，两端都会调用 CloseWithError 通知另一端，避免死锁。虽然示例中使用 time.Sleep 模拟慢速生产者，真实项目中应该用更可靠的数据源和取消信号。

另外，包装器链中如果某个中间层需要释放资源，一定要保证在所有数据写完后调用 Close。例如 gzip.Writer 的 Close 会写入 gzip 结尾标记；如果忘记调用，生成的压缩文件可能无法解压。可以用 defer 保证关闭，但要注意关闭顺序：应该从最外层包装器开始关闭，让每一层有机会刷新自身缓冲。