如何使用Golang分块上传大字符串或字节数组至AWS S3？

场景：从数据库导出特定数据并编码为JSON字符串，因数据量达数十亿字符，受内存限制无法一次性上传至AWS S3，需分块上传后生成完整可下载的JSON文件。

核心思路 #

利用AWS S3的**Multipart Upload（分块上传）**功能，将大数据拆分为多个小块依次上传，最后合并为单个文件。Golang可通过AWS SDK v2实现该流程，核心步骤为：初始化分块会话 → 分块上传数据 → 完成分块合并。

实现步骤及代码示例 #

1. 初始化S3客户端 #

先配置AWS凭证与区域，创建S3客户端实例：

import (
    "context"
    "fmt"
    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/s3"
    "github.com/aws/aws-sdk-go-v2/service/s3/types"
    "io"
    "strings"
)

func initS3Client(ctx context.Context) (*s3.Client, error) {
    cfg, err := config.LoadDefaultConfig(ctx, config.WithRegion("your-region"))
    if err != nil {
        return nil, fmt.Errorf("加载配置失败: %w", err)
    }
    return s3.NewFromConfig(cfg), nil
}

2. 创建分块上传会话 #

生成Upload ID，作为本次分块上传的唯一标识：

func createMultipartUpload(ctx context.Context, client *s3.Client, bucket, key string) (string, error) {
    resp, err := client.CreateMultipartUpload(ctx, &s3.CreateMultipartUploadInput{
        Bucket:      aws.String(bucket),
        Key:         aws.String(key),
        ContentType: aws.String("application/json"),
    })
    if err != nil {
        return "", fmt.Errorf("初始化分块上传失败: %w", err)
    }
    return *resp.UploadId, nil
}

3. 分块上传数据 #

将大字符串拆分为固定大小的块（S3要求非最后块最小5MB），逐个上传并记录每个块的ETag：

const chunkSize = 5 * 1024 * 1024 // 5MB分块大小

func uploadChunks(ctx context.Context, client *s3.Client, bucket, key, uploadID, largeStr string) ([]types.CompletedPart, error) {
    var completedParts []types.CompletedPart
    partNumber := 1
    strReader := strings.NewReader(largeStr)
    buf := make([]byte, chunkSize)

    for {
        n, err := strReader.Read(buf)
        if err != nil && err != io.EOF {
            return nil, fmt.Errorf("读取分块失败: %w", err)
        }
        if n == 0 {
            break
        }

        // 上传当前分块
        resp, err := client.UploadPart(ctx, &s3.UploadPartInput{
            Bucket:     aws.String(bucket),
            Key:        aws.String(key),
            UploadId:   aws.String(uploadID),
            PartNumber: aws.Int32(int32(partNumber)),
            Body:       strings.NewReader(string(buf[:n])),
        })
        if err != nil {
            return nil, fmt.Errorf("上传第%d块失败: %w", partNumber, err)
        }

        completedParts = append(completedParts, types.CompletedPart{
            ETag:       resp.ETag,
            PartNumber: aws.Int32(int32(partNumber)),
        })
        partNumber++
    }

    return completedParts, nil
}

4. 完成分块合并 #

所有分块上传完成后，调用接口合并为完整文件：

func completeMultipartUpload(ctx context.Context, client *s3.Client, bucket, key, uploadID string, parts []types.CompletedPart) error {
    _, err := client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
        Bucket:   aws.String(bucket),
        Key:      aws.String(key),
        UploadId: aws.String(uploadID),
        MultipartUpload: &types.CompletedMultipartUpload{
            Parts: parts,
        },
    })
    if err != nil {
        return fmt.Errorf("完成分块合并失败: %w", err)
    }
    return nil
}

5. 整合流程主函数 #

func main() {
    ctx := context.Background()
    bucket := "your-bucket-name"
    key := "large-data.json"
    largeStr := "your-extremely-large-json-string" // 实际场景建议从数据库流式获取

    client, err := initS3Client(ctx)
    if err != nil {
        panic(err)
    }

    uploadID, err := createMultipartUpload(ctx, client, bucket, key)
    if err != nil {
        panic(err)
    }

    parts, err := uploadChunks(ctx, client, bucket, key, uploadID, largeStr)
    if err != nil {
        // 上传失败时清理未完成的分块资源
        _, abortErr := client.AbortMultipartUpload(ctx, &s3.AbortMultipartUploadInput{
            Bucket:   aws.String(bucket),
            Key:      aws.String(key),
            UploadId: aws.String(uploadID),
        })
        if abortErr != nil {
            fmt.Printf("清理分块资源失败: %v\n", abortErr)
        }
        panic(err)
    }

    if err := completeMultipartUpload(ctx, client, bucket, key, uploadID, parts); err != nil {
        panic(err)
    }

    fmt.Println("分块上传完成")
}

关键优化建议 #

• 流式处理规避内存堆积：不要将所有数据库数据拼接成大字符串，改为从数据库流式查询+流式JSON编码，直接将数据流拆分为分块上传，彻底解决内存限制问题。
• 分块大小调优：S3允许分块范围为5MB-5GB，建议设置为10-100MB，平衡网络请求次数与单块上传稳定性。
• 并发上传提升效率：使用goroutine并发上传多个分块，但需控制并发数避免触发AWS限流。
• 断点续传支持：记录已成功上传的分块ETag与序号，若上传中断可从断点处继续，无需重新上传所有分块。
• JSON格式合法性保障：若生成数组型JSON，需确保分块后整体结构正确——比如第一块开头加[，中间块结尾加,，最后块结尾加]，避免合并后的JSON无效。
• 错误时的资源清理：上传失败务必调用AbortMultipartUpload清理未完成的分块，避免S3存储无效资源产生额外费用。

内容的提问来源于stack exchange，提问作者Jan Švábík