向SQS FIFO队列并发发消息遇限流错误的解决求助

问题根源 #

原错误是SDK重试模块的限流器可用令牌不足——1000个goroutine瞬间发起请求，加上重试请求，直接耗尽了SDK内置限流器的配额。而使用NoOpRateLimiter完全关闭SDK侧限流后，请求毫无节制地打向AWS，触发了SQS的API限流阈值，反而导致失败率上升。

有效解决方法 #

1. 控制goroutine并发数

不要一次性启动1000个goroutine，用带缓冲的channel做并发闸门，根据SQS配额调整并发量（比如FIFO队列默认全局TPS是3000，可设并发50-100）：

wg := sync.WaitGroup{}
messages := []string{}
// 并发控制闸门，限制同时运行的goroutine数量
semaphore := make(chan struct{}, 50)

for _, message := range messages {
    wg.Add(1)
    semaphore <- struct{}{}
    go func(msg string) {
        defer func() {
            wg.Done()
            <-semaphore
        }()
        // 在这里执行SQS SendMessage逻辑
    }(message)
}
wg.Wait()
close(semaphore)

2. 配置SDK自带令牌桶限流器

不要直接关闭限流器，而是配置符合SQS配额的令牌桶，平衡请求速率：

import (
    "context"
    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/aws/retry"
)

func loadSQSConfig(ctx context.Context) (aws.Config, error) {
    return config.LoadDefaultConfig(
        ctx,
        config.WithRetryer(func() aws.Retryer {
            // 令牌桶配置：每秒生成3000个令牌，桶容量3000（匹配FIFO默认TPS）
            rateLimiter := retry.NewTokenBucketRateLimiter(3000, 3000)
            return retry.NewStandard(func(options *retry.StandardOptions) {
                options.RateLimiter = rateLimiter
                // 合理设置重试次数，避免重试加剧限流
                options.MaxAttempts = 3
            })
        }),
    )
}

3. 使用批量发送API减少请求量

SQS的SendMessageBatch接口支持单次发送最多10条消息，能大幅降低请求次数，从根源减少限流概率：

import (
    "context"
    "fmt"
    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/service/sqs"
    "github.com/aws/aws-sdk-go-v2/service/sqs/types"
)

func sendBatch(client *sqs.Client, queueURL string, batch []string) error {
    entries := make([]types.SendMessageBatchRequestEntry, 0, len(batch))
    for idx, msg := range batch {
        entries = append(entries, types.SendMessageBatchRequestEntry{
            Id:                      aws.String(fmt.Sprintf("batch-msg-%d", idx)),
            MessageBody:             aws.String(msg),
            MessageGroupId:          aws.String("your-business-group-id"), // FIFO必填
            MessageDeduplicationId:  aws.String(fmt.Sprintf("dedup-%d", idx)), // 可选，避免重复消息
        })
    }

    _, err := client.SendMessageBatch(context.TODO(), &sqs.SendMessageBatchInput{
        QueueUrl: aws.String(queueURL),
        Entries:  entries,
    })
    return err
}

4. 结合并发控制与批量发送

将两种方式结合，既控制并发数，又减少请求次数，是最高效的方案：

wg := sync.WaitGroup{}
messages := []string{}
semaphore := make(chan struct{}, 10) // 批量发送的并发数
batchSize := 10

// 拆分消息为批量组
var batches [][]string
for i := 0; i < len(messages); i += batchSize {
    end := i + batchSize
    if end > len(messages) {
        end = len(messages)
    }
    batches = append(batches, messages[i:end])
}

for _, batch := range batches {
    wg.Add(1)
    semaphore <- struct{}{}
    go func(b []string) {
        defer func() {
            wg.Done()
            <-semaphore
        }()
        if err := sendBatch(client, queueURL, b); err != nil {
            // 错误处理：记录日志、重试当前批次等
            fmt.Printf("Batch send failed: %v\n", err)
        }
    }(batch)
}
wg.Wait()
close(semaphore)

内容的提问来源于stack exchange，提问作者shekwo