使用s3gof3r实现高性能Amazon S3大文件传输

s3gof3r是一个专为高性能Amazon S3大文件传输设计的Go语言库，它通过并发上传/下载、内存优化和智能分块等机制，显著提高了大文件传输速度。

安装

go get github.com/rlmcpherson/s3gof3r

基本配置

首先需要配置AWS凭证：

import (
    "github.com/rlmcpherson/s3gof3r"
)

func getS3Config() *s3gof3r.Config {
    keys := s3gof3r.Keys{
        AccessKey: "YOUR_AWS_ACCESS_KEY",
        SecretKey: "YOUR_AWS_SECRET_KEY",
    }
    
    return s3gof3r.NewConfig(
        s3gof3r.WithLogger(s3gof3r.DefaultLogger), // 可选日志
        s3gof3r.WithConcurrency(10),               // 并发数
        s3gof3r.WithPartSize(20 * 1024 * 1024),   // 分块大小20MB
        s3gof3r.WithScheme("https"),              // 使用HTTPS
        s3gof3r.WithRegion("us-west-1"),          // 区域
    )
}

文件上传

func uploadFile(bucketName, s3Path, localPath string) error {
    // 初始化S3 bucket
    s3 := s3gof3r.New("", getS3Config())
    bucket := s3.Bucket(bucketName)
    
    // 打开本地文件
    file, err := os.Open(localPath)
    if err != nil {
        return fmt.Errorf("failed to open file: %v", err)
    }
    defer file.Close()
    
    // 创建S3写入器
    writer, err := bucket.PutWriter(s3Path, nil, nil)
    if err != nil {
        return fmt.Errorf("failed to create S3 writer: %v", err)
    }
    defer writer.Close()
    
    // 复制文件内容
    if _, err = io.Copy(writer, file); err != nil {
        return fmt.Errorf("failed to copy file to S3: %v", err)
    }
    
    return nil
}

文件下载

func downloadFile(bucketName, s3Path, localPath string) error {
    // 初始化S3 bucket
    s3 := s3gof3r.New("", getS3Config())
    bucket := s3.Bucket(bucketName)
    
    // 创建本地文件
    file, err := os.Create(localPath)
    if err != nil {
        return fmt.Errorf("failed to create local file: %v", err)
    }
    defer file.Close()
    
    // 创建S3读取器
    reader, _, err := bucket.GetReader(s3Path, nil)
    if err != nil {
        return fmt.Errorf("failed to create S3 reader: %v", err)
    }
    defer reader.Close()
    
    // 复制文件内容
    if _, err = io.Copy(file, reader); err != nil {
        return fmt.Errorf("failed to copy S3 file to local: %v", err)
    }
    
    return nil
}

高级功能

1. 进度监控

type progressReader struct {
    io.Reader
    total    int64
    current  int64
}

func (r *progressReader) Read(p []byte) (n int, err error) {
    n, err = r.Reader.Read(p)
    r.current += int64(n)
    fmt.Printf("\rDownload progress: %.2f%%", float64(r.current)/float64(r.total)*100)
    return
}

func downloadWithProgress(bucketName, s3Path, localPath string) error {
    // ...前面的初始化代码...
    
    // 获取文件大小
    head, err := bucket.Head(s3Path)
    if err != nil {
        return err
    }
    
    reader, _, err := bucket.GetReader(s3Path, nil)
    if err != nil {
        return err
    }
    defer reader.Close()
    
    progress := &progressReader{
        Reader: reader,
        total:  head.Size,
    }
    
    if _, err = io.Copy(file, progress); err != nil {
        return err
    }
    
    fmt.Println("\nDownload completed!")
    return nil
}

2. 多部分上传控制

func uploadWithCustomParts(bucketName, s3Path, localPath string) error {
    config := getS3Config()
    config.PartSize = 50 * 1024 * 1024 // 50MB分块
    config.Concurrency = 5             // 5个并发
    
    s3 := s3gof3r.New("", config)
    bucket := s3.Bucket(bucketName)
    
    file, err := os.Open(localPath)
    if err != nil {
        return err
    }
    defer file.Close()
    
    writer, err := bucket.PutWriter(s3Path, nil, nil)
    if err != nil {
        return err
    }
    defer writer.Close()
    
    _, err = io.Copy(writer, file)
    return err
}

性能优化建议

1. 调整分块大小：对于非常大的文件(>1GB)，增加分块大小(如50-100MB)可以提高性能
2. 增加并发数：根据网络带宽和CPU能力适当增加并发数(通常10-20)
3. 使用EC2实例：在AWS EC2上运行传输，特别是与S3在同一区域
4. 启用压缩：对于可压缩文件类型，可以启用gzip压缩
5. 批量操作：对于多个小文件，考虑先打包再传输

错误处理

func safeUpload(bucketName, s3Path, localPath string) error {
    s3 := s3gof3r.New("", getS3Config())
    bucket := s3.Bucket(bucketName)
    
    file, err := os.Open(localPath)
    if err != nil {
        return fmt.Errorf("file open error: %v", err)
    }
    defer file.Close()
    
    writer, err := bucket.PutWriter(s3Path, nil, nil)
    if err != nil {
        return fmt.Errorf("S3 writer creation error: %v", err)
    }
    
    // 确保在出错时中止上传
    success := false
    defer func() {
        writer.Close()
        if !success {
            bucket.Delete(s3Path) // 删除不完整的上传
        }
    }()
    
    if _, err = io.Copy(writer, file); err != nil {
        return fmt.Errorf("upload error: %v", err)
    }
    
    success = true
    return nil
}

s3gof3r相比标准AWS SDK在大文件传输场景下性能显著提升，特别适合TB级数据迁移、备份恢复等场景。通过合理配置参数，可以充分利用带宽资源，实现最快的传输速度。