Go语言中通过Channel高效传输Zlib压缩数据流的教程

func Compress(r io.Reader) (<-chan byte) {
    c := make(chan byte)
    go func(){
        var wBuff bytes.Buffer // 存储压缩数据的缓冲区
        rBuff := make([]byte, 1024)
        writer := zlib.NewWriter(&wBuff) // zlib writer将数据写入wBuff
        for {
            n, err := r.Read(rBuff)
            if err != nil && err != io.EOF { panic(err) }
            if n == 0 { break }
            writer.Write(rBuff[:n]) // 压缩并写入wBuff
            // 问题：如何从wBuff中获取已压缩的数据并发送到channel c？
            // wBuff会累积所有压缩数据，无法实时获取
        }
        writer.Close() // 关闭writer，确保所有数据被flush
        // 此时wBuff才包含完整的压缩数据
        // 如何将wBuff中的数据分块发送？
        close(c)
    }()
    return c
}

type BytesWithError struct {
    Bytes []byte
    Err   error
}

type ChanWriter chan BytesWithError

// Write 方法将接收到的字节切片发送到其内部的channel。
// 注意：为了避免并发修改共享切片的问题，通常会发送切片的副本。
func (cw ChanWriter) Write(p []byte) (n int, err error) {
    // 创建p的副本，以避免发送后p被修改导致的数据竞争问题
    bufCopy := make([]byte, len(p))
    copy(bufCopy, p)
    cw <- BytesWithError{Bytes: bufCopy}
    return len(p), nil
}

package main

import (
    "bytes"
    "compress/zlib"
    "io"
    "log"
)

// BytesWithError 结构体用于通过channel传输字节切片和可能的错误
type BytesWithError struct {
    Bytes []byte
    Err   error
}

// ChanWriter 类型实现了 io.Writer 接口，将数据写入到其内部的channel
type ChanWriter chan BytesWithError

// Write 方法将接收到的字节切片发送到其内部的channel。
// 为了避免并发修改共享切片的问题，这里发送切片的副本。
func (cw ChanWriter) Write(p []byte) (n int, err error) {
    // 创建p的副本，以避免发送后p被修改导致的数据竞争问题
    bufCopy := make([]byte, len(p))
    copy(bufCopy, p)
    cw <- BytesWithError{Bytes: bufCopy}
    return len(p), nil
}

// Compress 函数通过channel传输zlib压缩后的数据流
func Compress(r io.Reader) <-chan BytesWithError {
    outputChan := make(chan BytesWithError)

    go func() {
        defer close(outputChan) // 确保channel在goroutine结束时关闭

        // 创建一个ChanWriter实例，它将把数据发送到outputChan
        cw := ChanWriter(outputChan)

        // zlib.NewWriter 将压缩数据写入到 cw (它实现了 io.Writer)
        zlibWriter := zlib.NewWriter(cw)
        defer func() {
            if err := zlibWriter.Close(); err != nil {
                // 如果关闭zlibWriter时发生错误，发送错误到channel
                outputChan <- BytesWithError{Err: err}
            }
        }()

        // 从输入io.Reader读取数据并写入zlibWriter进行压缩
        // zlibWriter 会将压缩后的数据通过 cw (即 outputChan) 发送出去
        _, err := io.Copy(zlibWriter, r)
        if err != nil {
            // 如果io.Copy过程中发生错误，发送错误到channel
            outputChan <- BytesWithError{Err: err}
        }
    }()

    return outputChan
}

func main() {
    // 示例：一个模拟的io.Reader作为输入源
    originalData := "This is a long string that needs to be compressed and streamed. " +
        "We will repeat it multiple times to simulate a larger data source. " +
        "Go channels are excellent for this kind of concurrent data processing. "

    var inputBuf bytes.Buffer
    for i := 0; i < 100; i++ { // 模拟大量数据
        inputBuf.WriteString(originalData)
    }

    reader := bytes.NewReader(inputBuf.Bytes())

    // 调用Compress函数，获取一个接收压缩数据的channel
    compressedStream := Compress(reader)

    // 创建一个bytes.Buffer来收集所有压缩后的数据，以便验证
    var receivedCompressedData bytes.Buffer

    // 从channel中读取压缩数据
    for chunk := range compressedStream {
        if chunk.Err != nil {
            log.Fatalf("Error during compression: %v", chunk.Err)
        }
        if chunk.Bytes !=