怎样用Golang实现高效文件压缩传输 集成zstd与snappy流式压缩

package main

import (
    "fmt"
    "io"
    "log"
    "net"
    "os"
    "time"

    "github.com/golang/snappy/snappy" // Snappy
    "github.com/klauspost/compress/zstd" // Zstandard
)

// 假设我们有一个文件要传输
const (
    testFileName = "large_test_file.txt"
    serverAddr   = "localhost:8080"
)

// createDummyFile 创建一个测试文件
func createDummyFile(name string, size int64) error {
    f, err := os.Create(name)
    if err != nil {
        return err
    }
    defer f.Close()

    // 写入一些重复数据，便于压缩
    data := make([]byte, 1024)
    for i := 0; i < len(data); i++ {
        data[i] = byte(i % 256)
    }

    written := int64(0)
    for written < size {
        n, err := f.Write(data)
        if err != nil {
            return err
        }
        written += int64(n)
    }
    fmt.Printf("Created dummy file %s with size %d bytes\n", name, size)
    return nil
}

// sendFileStream 客户端发送文件函数
// compressorType: "zstd" or "snappy"
func sendFileStream(filePath string, conn net.Conn, compressorType string) error {
    file, err := os.Open(filePath)
    if err != nil {
        return fmt.Errorf("failed to open file: %w", err)
    }
    defer file.Close()

    var compressedWriter io.WriteCloser
    switch compressorType {
    case "zstd":
        // zstd.NewWriter 默认提供了较好的平衡
        w, err := zstd.NewWriter(conn)
        if err != nil {
            return fmt.Errorf("failed to create zstd writer: %w", err)
        }
        compressedWriter = w
        fmt.Printf("Sending file %s using Zstandard compression...\n", filePath)
    case "snappy":
        // snappy.NewBufferedWriter 提供了缓冲，性能更好
        compressedWriter = snappy.NewBufferedWriter(conn)
        fmt.Printf("Sending file %s using Snappy compression...\n\n", filePath)
    default:
        return fmt.Errorf("unsupported compressor type: %s", compressorType)
    }
    defer compressedWriter.Close() // 确保关闭压缩器，刷新所有数据

    start := time.Now()
    n, err := io.Copy(compressedWriter, file)
    if err != nil {
        return fmt.Errorf("failed to copy file to compressed writer: %w", err)
    }
    duration := time.Since(start)
    fmt.Printf("Sent %d bytes (compressed) in %s\n", n, duration)
    return nil
}

// receiveFileStream 服务器接收文件函数
// compressorType: "zstd" or "snappy"
func receiveFileStream(conn net.Conn, outputFilePath string, compressorType string) error {
    outputFile, err := os.Create(outputFilePath)
    if err != nil {
        return fmt.Errorf("failed to create output file: %w", err)
    }
    defer outputFile.Close()

    var decompressedReader io.Reader
    switch compressorType {
    case "zstd":
        r, err := zstd.NewReader(conn)
        if err != nil {
            return fmt.Errorf("failed to create zstd reader: %w", err)
        }
        decompressedReader = r
        defer r.Close() // zstd reader需要关闭
        fmt.Printf("Receiving file using Zstandard decompression...\n")
    case "snappy":
        decompressedReader = snappy.NewReader(conn)
        fmt.Printf("Receiving file using Snappy decompression...\n")
    default:
        return fmt.Errorf("unsupported compressor type: %s", compressorType)
    }

    start := time.Now()
    n, err := io.Copy(outputFile, decompressedReader)
    if err != nil {
        return fmt.Errorf("failed to copy decompressed data to file: %w", err)
    }
    duration := time.Since(start)
    fmt.Printf("Received and decompressed %d bytes in %s\n", n, duration)
    return nil
}

func main() {
    // 创建一个大文件用于测试
    if err := createDummyFile(testFileName, 50*1024*1024); err != nil { // 50MB
        log.Fatalf("Error creating dummy file: %v", err)
    }
    defer os.Remove(testFileName) // 清理

    // 启动服务器
    go func() {
        listener, err := net.Listen("tcp", serverAddr)
        if err != nil {
            log.Fatalf("Server listen error: %v", err)
        }
        defer listener.Close()
        fmt.Println("Server listening on", serverAddr)

        conn, err := listener.Accept()
        if err != nil {
            log.Printf("Server accept error: %v", err)
            return
        }
        defer conn.Close()
        fmt.Println("Server accepted connection from", conn.RemoteAddr())

        // 接收文件，这里我们假设服务器知道客户端是用zstd发送的
        if err := receiveFileStream(conn, "received_zstd_file.txt", "zstd"); err != nil {
            log.Printf("Server receive file error (zstd): %v", err)
        }
        fmt.Println("Zstd file received.")

        // 再次等待连接，接收snappy文件
        conn2, err := listener.Accept()
        if err != nil {
            log.Printf("Server accept error 2: %v", err)
            return
        }
        defer conn2.Close()
        fmt.Println("Server accepted second connection from", conn2.RemoteAddr())
        if err := receiveFileStream(conn2, "received_snappy_file.txt", "snappy"); err != nil {
            log.Printf("Server receive file error (snappy): %v", err)
        }
        fmt.Println("Snappy file received.")

        // 验证文件是否一致 (省略具体校验逻辑，仅作示意)
        fmt.Println("File transfer complete. Check 'received_zstd_file.txt' and 'received_snappy_file.txt'")
    }()

    time.Sleep(time.Second) // 等待服务器启动

    // 客户端发送zstd压缩的文件
    conn1, err := net.Dial("tcp", serverAddr)
    if err != nil {
        log.Fatalf("Client dial error (zstd): %v", err)
    }
    defer conn1.Close()
    if err := sendFileStream(testFileName, conn1, "zstd"); err != nil {
        log.Fatalf("Client send file error (zstd): %v", err)
    }
    fmt.Println("Zstd file sent.")
    time.Sleep(time.Second) // 给服务器一点处理时间

    // 客户端发送snappy压缩的文件
    conn2, err := net.Dial("tcp", serverAddr)
    if err != nil {
        log.Fatalf("Client dial error (snappy): %v", err)
    }
    defer conn2.Close()
    if err := sendFileStream(testFileName, conn2, "snappy"); err != nil {
        log.Fatalf("Client send file error (snappy): %v", err)
    }
    fmt.Println("Snappy file sent.")

    // 等待服务器处理完毕
    time.Sleep(2 * time.Second)
}