如何处理调用限频200次/分钟API的AWS Lambda限流问题

1. 在Lambda中实现分布式令牌桶限流 #

因为SQS触发的Lambda是无状态且可能并发执行的，直接调用API很容易突破200次/分钟的限制。可以基于DynamoDB实现分布式令牌桶来统一控制API调用速率：

• 每次Lambda执行时，先从DynamoDB读取当前剩余令牌数和上次令牌刷新时间
• 如果距离上次刷新已超过1分钟，重置令牌数为200，并更新刷新时间戳
• 若剩余令牌>0，消耗1个令牌后调用API；若令牌不足，将消息重新放回SQS（设置合适的可见性超时，比如30秒），等待后续重试
• 示例伪代码：

import boto3
from datetime import datetime, timedelta

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('RateLimitTokens')

def get_token():
    response = table.get_item(Key={'id': 'api_rate_limit'})
    item = response.get('Item', {})
    now = datetime.utcnow()
    last_refresh = item.get('last_refresh', datetime.min)
    
    # 每分钟重置令牌
    if now - last_refresh > timedelta(minutes=1):
        table.put_item(Item={'id': 'api_rate_limit', 'tokens': 200, 'last_refresh': now})
        return True
    # 尝试获取令牌
    elif item.get('tokens', 0) > 0:
        table.update_item(
            Key={'id': 'api_rate_limit'},
            UpdateExpression='SET tokens = tokens - :val',
            ExpressionAttributeValues={':val': 1},
            ConditionExpression='tokens > :zero',
            ExpressionAttributeValues={':zero': 0}
        )
        return True
    return False

def lambda_handler(event, context):
    if not get_token():
        # 令牌不足，将消息放回SQS，设置可见性超时30秒
        sqs = boto3.client('sqs')
        for record in event['Records']:
            sqs.change_message_visibility(
                QueueUrl='YOUR_SQS_QUEUE_URL',
                ReceiptHandle=record['receiptHandle'],
                VisibilityTimeout=30
            )
        return {'statusCode': 429, 'body': 'Rate limit exceeded, retrying later'}
    
    # 调用目标API逻辑
    # ...

2. 严格控制Lambda并发与SQS触发参数 #

调整Lambda和SQS的配置，从触发源头限制并发请求数：

• 设置Lambda的预留并发数：按API速率倒推，假设每个Lambda单次调用仅发起1次API请求，那么预留并发设为3（3次并发×60秒/分钟=180次/分钟，接近200的限额，留有余量）
• 配置SQS触发Lambda的参数：
  • Batch Size设为1，确保每条消息对应一次API调用
  • MaximumBatchingWindowInSeconds设为0，立即触发Lambda
  • 将SQS的可见性超时设为Lambda最大执行时间的3倍（比如Lambda最多跑10秒，就设30秒），避免消息被重复拉取

3. 用EventBridge Pipes实现流量控制 #

EventBridge Pipes支持直接对SQS到Lambda的转发设置速率限制，无需额外代码：

• 创建一个Pipe，源为你的SQS队列，目标为Lambda函数
• 在Pipe的流量控制配置中，设置速率限制为200次/分钟
• Pipe会自动拦截并限流，确保Lambda调用API的总次数不超过限额

4. Step Functions编排限流流程 #

用Step Functions的速率控制能力替代直接的SQS-Lambda触发：

• 创建状态机，添加一个Task节点调用Lambda，同时在状态机的Rate参数中设置200/分钟
• 将SQS队列配置为触发该状态机
• 状态机会自动调度请求，确保每分钟仅执行200次Lambda调用，完全匹配API的速率限制

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