Golang中如何高效使用io.Pipe实现HTTP多部分文件上传

以下是针对您问题的专业解答，包含代码示例：

问题1：如何将uploadToS3中的错误通知给flushToWriter

使用context.Context来协调goroutine的取消：

func uploadToS3(URL string, traceFile TraceFile) error {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    fileSize, err := util.GetFileSize(traceFile.systemPath)
    if err != nil {
        return err
    }
    
    contentLength := emptyMultipartSize("file", traceFile.fileName) + fileSize
    readBody, writeBody := io.Pipe()
    defer readBody.Close()

    form := multipart.NewWriter(writeBody)
    errChan := make(chan error, 1)
    
    // 传递context
    go flushToWriter(ctx, errChan, writeBody, traceFile)
    
    req, err := http.NewRequestWithContext(ctx, http.MethodPut, URL, readBody)
    if err != nil {
        cancel() // 取消context
        return err
    }
    
    req.Header.Set("Content-Type", form.FormDataContentType())
    if contentLength > 0 {
        req.ContentLength = contentLength
    }
    
    res, err := http.DefaultClient.Do(req)
    if err != nil {
        log.Error("Uploading failed due to", err)
        cancel() // 通知flushToWriter停止
        <-errChan
        return err
    }
    defer res.Body.Close()
    
    log.Info("Response from s3 ", res)
    return <-errChan
}

func flushToWriter(ctx context.Context, errChan chan error, writeBody *io.PipeWriter, traceFile TraceFile) {
    form := multipart.NewWriter(writeBody)
    defer writeBody.Close()

    filePart, err := form.CreateFormFile("file", traceFile.fileName)
    if err != nil {
        select {
        case errChan <- err:
        case <-ctx.Done():
        }
        return
    }

    err = copyToWriter(ctx, traceFile.systemPath, filePart)
    if err != nil {
        select {
        case errChan <- err:
        case <-ctx.Done():
        }
        return
    }

    select {
    case errChan <- form.Close():
    case <-ctx.Done():
        errChan <- ctx.Err()
    }
}

func copyToWriter(ctx context.Context, path string, filePart io.Writer) error {
    file, err := os.Open(path)
    if err != nil {
        return err
    }
    defer file.Close()

    fi, err := file.Stat()
    if err != nil {
        return err
    }
    
    // 使用带context的Copy
    _, err = io.Copy(filePart, file)
    return err
}

问题2：当flushToWriter出现错误时如何停止请求

在flushToWriter中关闭管道writer会中断HTTP请求：

func flushToWriter(ctx context.Context, errChan chan error, writeBody *io.PipeWriter, traceFile TraceFile) {
    form := multipart.NewWriter(writeBody)
    defer func() {
        // 确保管道关闭以中断HTTP请求
        writeBody.CloseWithError(err)
    }()

    filePart, err := form.CreateFormFile("file", traceFile.fileName)
    if err != nil {
        errChan <- err
        return
    }

    err = copyToWriter(ctx, traceFile.systemPath, filePart)
    if err != nil {
        errChan <- err
        return
    }

    err = form.Close()
    if err != nil {
        errChan <- err
        return
    }
    
    errChan <- nil
}

问题3：管道使用是否正确

您的管道使用基本正确，但可以优化错误处理：

func uploadToS3(URL string, traceFile TraceFile) error {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    fileSize, err := util.GetFileSize(traceFile.systemPath)
    if err != nil {
        return err
    }
    
    contentLength := emptyMultipartSize("file", traceFile.fileName) + fileSize
    readBody, writeBody := io.Pipe()

    errChan := make(chan error, 1)
    go func() {
        defer close(errChan)
        defer readBody.Close()
        
        form := multipart.NewWriter(writeBody)
        defer writeBody.Close()

        filePart, err := form.CreateFormFile("file", traceFile.fileName)
        if err != nil {
            errChan <- err
            return
        }

        file, err := os.Open(traceFile.systemPath)
        if err != nil {
            errChan <- err
            return
        }
        defer file.Close()

        // 流式复制，避免内存占用
        _, err = io.Copy(filePart, file)
        if err != nil {
            errChan <- err
            return
        }

        err = form.Close()
        errChan <- err
    }()

    req, err := http.NewRequestWithContext(ctx, http.MethodPut, URL, readBody)
    if err != nil {
        cancel()
        return err
    }
    
    req.Header.Set("Content-Type", "multipart/form-data")
    if contentLength > 0 {
        req.ContentLength = contentLength
    }
    
    res, err := http.DefaultClient.Do(req)
    if err != nil {
        cancel()
        // 等待goroutine完成或超时
        select {
        case <-errChan:
        case <-time.After(5 * time.Second):
        }
        return err
    }
    defer res.Body.Close()

    log.Info("Response from s3 ", res)
    
    // 检查goroutine中的错误
    if err := <-errChan; err != nil {
        return fmt.Errorf("upload failed: %w", err)
    }
    
    return nil
}

这些修改确保了：

1. 使用context进行协调取消
2. 管道错误正确传播
3. 资源正确清理
4. 内存高效的流式处理