如何用Golang优化云存储操作 实现S3高性能客户端

s3manager

s3manager.Uploader

import (
    "context"
    "fmt"
    "io"
    "os"
    "time"

    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/feature/s3/manager"
    "github.com/aws/aws-sdk-go-v2/service/s3"
)

// Assume cfg is your aws.Config loaded with credentials and region
func uploadFileToS3(ctx context.Context, s3Client *s3.Client, bucket, key, filePath string) error {
    file, err := os.Open(filePath)
    if err != nil {
        return fmt.Errorf("failed to open file %s: %w", filePath, err)
    }
    defer file.Close()

    uploader := manager.NewUploader(s3Client, func(u *manager.Uploader) {
        u.PartSize = 64 * 1024 * 1024 // 64MB per part, adjust based on network
        u.Concurrency = 10            // Upload 10 parts concurrently
        u.BufferProvider = manager.NewBufferedReadFromProvider(32 * 1024 * 1024) // Buffer for each part
    })

    fmt.Printf("Starting multipart upload for %s to s3://%s/%s\n", filePath, bucket, key)
    start := time.Now()

    _, err = uploader.Upload(ctx, &s3.PutObjectInput{
        Bucket: aws.String(bucket),
        Key:    aws.String(key),
        Body:   file,
    })

    if err != nil {
        return fmt.Errorf("failed to upload file %s: %w", filePath, err)
    }

    fmt.Printf("Successfully uploaded %s in %s\n", filePath, time.Since(start))
    return nil
}

BufferProvider

s3manager.Downloader

io.WriterAt

// Assume cfg is your aws.Config loaded with credentials and region
func downloadFileFromS3(ctx context.Context, s3Client *s3.Client, bucket, key, outputPath string) error {
    file, err := os.Create(outputPath)
    if err != nil {
        return fmt.Errorf("failed to create file %s: %w", outputPath, err)
    }
    defer file.Close()

    downloader := manager.NewDownloader(s3Client, func(d *manager.Downloader) {
        d.PartSize = 64 * 1024 * 1024 // 64MB per part, same logic as upload
        d.Concurrency = 10            // Download 10 parts concurrently
    })

    fmt.Printf("Starting multipart download for s3://%s/%s to %s\n", bucket, key, outputPath)
    start := time.Now()

    // Downloader writes to io.WriterAt
    numBytes, err := downloader.Download(ctx, file, &s3.GetObjectInput{
        Bucket: aws.String(bucket),
        Key:    aws.String(key),
    })

    if err != nil {
        return fmt.Errorf("failed to download file %s: %w", outputPath, err)
    }

    fmt.Printf("Successfully downloaded %d bytes to %s in %s\n", numBytes, outputPath, time.Since(start))
    return nil
}

net/http

http.DefaultTransport

http.Client

http.Transport

MaxIdleConns

IdleConnTimeout

import (
    "net/http"
    "time"

    "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"
)

func createOptimizedS3Client(ctx context.Context, region string) (*s3.Client, error) {
    // Custom HTTP client with optimized transport settings
    tr := &http.Transport{
        MaxIdleConns:        100,              // Total maximum idle connections across all hosts
        MaxIdleConnsPerHost: 20,               // Maximum idle connections to a single host (S3 endpoint)
        IdleConnTimeout:     90 * time.Second, // How long an idle connection is kept alive
        DisableKeepAlives:   false,            // Ensure keep-alives are enabled
        // You might also consider ResponseHeaderTimeout, ExpectContinueTimeout for specific scenarios
    }
    httpClient := &http.Client{
        Transport: tr,
        Timeout:   30 * time.Second, // Overall request timeout
    }

    cfg, err := config.LoadDefaultAWSConfig(ctx, config.WithRegion(region))
    if err != nil {
        return nil, fmt.Errorf("failed to load AWS config: %w", err)
    }

    // Override the default HTTP client in the AWS SDK
    s3Client := s3.NewFromConfig(cfg, func(o *s3.Options) {
        o.HTTPClient = httpClient
    })

    return s3Client, nil
}

// Simplified worker pool example for processing S3 objects
func processS3ObjectsConcurrently(ctx context.Context, s3Client *s3.Client, bucket string, objectKeys []string, numWorkers int) {
    jobs := make(chan string, len(objectKeys))
    results := make(chan error, len(objectKeys))

    // Start workers
    for i := 0; i < numWorkers; i++ {
        go func() {
            for key := range jobs {
                // Simulate an S3 operation, e.g., get object metadata
                _, err := s3Client.HeadObject(ctx, &s3.HeadObjectInput{
                    Bucket: aws.String(bucket),
                    Key:    aws.String(key),
                })
                if err != nil {
                    results <- fmt.Errorf("failed to head object %s: %w", key, err)
                    continue
                }
                results <- nil // Success
            }
        }()
    }

    // Send jobs
    for _, key := range objectKeys {
        jobs <- key
    }
    close(jobs)

    // Collect results
    for i := 0; i < len(objectKeys); i++ {
        err := <-results
        if err != nil {
            fmt.Printf("Error processing object: %v\n", err)
        }
    }
    fmt.Println("All objects processed.")
}

context

context.Context

// Example with timeout context
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel() // Always call cancel to release resources

// Use this ctx in your S3 operations
_, err := s3Client.GetObject(ctx, &s3.GetObjectInput{...})
if err != nil {
    if errors.Is(ctx.Err(), context.DeadlineExceeded) {
        fmt.Println("S3 GetObject timed out!")
    } else {
        fmt.Printf("S3 GetObject failed: %v\n", err)
    }
}

[]byte

sync.Pool