Go语言中高效处理大尺寸数据流与HTTP请求

package main

import (
    "bytes"
    "fmt"
    "time"
)

func main() {
    // 假设我们预期处理的数据大小约为100MB
    largeDataSize := 100 * 1024 * 1024 // 100 MB

    fmt.Println("--- 预分配缓冲区示例 ---")
    // 方法一：使用make([]byte, 0, capacity)预分配
    // 创建一个初始长度为0，但容量为largeDataSize的字节切片
    preAllocatedBuffer := bytes.NewBuffer(make([]byte, 0, largeDataSize))

    fmt.Printf("预分配缓冲区初始容量: %d MB\n", preAllocatedBuffer.Cap()/(1024*1024))
    start := time.Now()
    // 模拟写入100MB数据
    for i := 0; i < largeDataSize; i++ {
        preAllocatedBuffer.WriteByte('a')
    }
    duration := time.Since(start)
    fmt.Printf("写入 %d MB 数据耗时 (预分配): %v\n", largeDataSize/(1024*1024), duration)
    fmt.Printf("预分配缓冲区最终容量: %d MB\n", preAllocatedBuffer.Cap()/(1024*1024))
    // 重置缓冲区以便后续操作，但容量不变
    preAllocatedBuffer.Reset()


    fmt.Println("\n--- 非预分配缓冲区示例 ---")
    // 方法二：不预分配，让bytes.Buffer自动扩容
    unAllocatedBuffer := &bytes.Buffer{} // 默认初始容量很小
    fmt.Printf("非预分配缓冲区初始容量: %d B\n", unAllocatedBuffer.Cap()) // 初始容量通常是0或很小
    start = time.Now()
    // 模拟写入100MB数据
    for i := 0; i < largeDataSize; i++ {
        unAllocatedBuffer.WriteByte('a')
    }
    duration = time.Since(start)
    fmt.Printf("写入 %d MB 数据耗时 (非预分配): %v\n", largeDataSize/(1024*1024), duration)
    fmt.Printf("非预分配缓冲区最终容量: %d MB\n", unAllocatedBuffer.Cap()/(1024*1024))
}

package main

import (
    "fmt"
    "io"
    "log"
    "net/http"
    "time"
)

// downloadAndUploadStream 模拟从源URL下载数据并流式上传到目标URL
func downloadAndUploadStream(downloadURL, uploadURL string) error {
    log.Printf("开始从 %s 下载...", downloadURL)
    // 1. 发起下载请求
    resp, err := http.Get(downloadURL)
    if err != nil {
        return fmt.Errorf("下载请求失败: %w", err)
    }
    defer resp.Body.Close() // 确保关闭响应体

    if resp.StatusCode != http.StatusOK {
        return fmt.Errorf("下载文件HTTP状态码非200: %s", resp.Status)
    }

    // 2. 创建一个管道，用于连接下载流和上传流
    // pr (PipeReader) 实现了 io.Reader 接口
    // pw (PipeWriter) 实现了 io.Writer 接口
    pr, pw := io.Pipe()

    // 3. 在一个goroutine中将下载的响应体写入管道的写入端
    go func() {
        defer pw.Close() // 确保管道写入端最终关闭，这会通知读取端已到达EOF
        log.Printf("开始将下载数据写入管道...")
        _, copyErr := io.Copy(pw, resp.Body)
        if copyErr != nil && copyErr != io.EOF { // io.EOF通常是正常结束
            log.Printf("写入管道失败: %v", copyErr)
        }
        log.Printf("下载数据写入管道完成。")
    }()

    // 4. 创建上传请求，请求体直接使用管道的读取端
    // 这使得HTTP客户端可以在下载数据写入管道的同时，从管道读取数据并上传
    req, err := http.NewRequest(http.MethodPost, uploadURL, pr)
    if err != nil {
        return fmt.Errorf("创建上传请求失败: %w", err)
    }

    // 如果Content-Length已知，设置它有助于服务器接收
    // 但对于流式上传，通常不设置或设置为-1，让客户端使用分块传输编码(chunked transfer encoding)
    // if resp.ContentLength > 0 {
    //  req.ContentLength = resp.ContentLength
    // }
    // req.Header.Set("Content-Type", "application/octet-stream") // 根据实际情况设置

    log.Printf("开始流式上传到 %s...", uploadURL)
    client := &http.Client{
        Timeout: 300 * time.Second, // 设置一个较长的超时时间
    }
    uploadResp, err := client.Do(req)
    if err != nil {
        // 如果上传失败，需要关闭管道的读取端以释放资源
        pr.CloseWithError(fmt.Errorf("上传请求失败: %w", err))
        return fmt.Errorf("上传文件失败: %w", err)
    }
    defer uploadResp.Body.Close() // 确保关闭上传响应体

    if uploadResp.StatusCode != http.StatusOK {
        return fmt.Errorf("上传文件HTTP状态码非200: %s", uploadResp.Status)
    }

    log.Println("文件流式下载和上传成功！")
    return nil
}

func main() {
    // 替换为实际的下载和上传URL以进行测试
    // downloadURL := "http://speedtest.tele2.net/100MB.zip" // 示例：一个100MB的测试文件
    // uploadURL := "http://your-upload-server.com/upload" // 替换为你的上传接口

    // if err := downloadAndUploadStream(downloadURL, uploadURL); err != nil {
    //  log.Fatalf("操作失败: %v", err)
    // } else {
    //  fmt.Println("流式传输演示完成。")
    // }

    fmt.Println("此示例展示了流式处理的概念，需要真实的URL才能运行。")
    fmt.Println("请自行替换 `downloadURL` 和 `uploadURL` 进行测试。")
}