如何配置AWS FIFO SQS队列实现5秒内多条消息合并为单个事件？

要实现将5秒内发送的多条FIFO SQS消息合并为一个事件触发Lambda，核心是利用Lambda的批量触发配置结合FIFO队列的消息分组特性，具体步骤如下：

1. 配置FIFO SQS队列 #

• 创建或修改FIFO队列时，确保启用以下特性：
  • 内容去重（可选但推荐）：开启Content-Based Deduplication避免重复消息
  • 保留默认的FIFO Queue标识，确保队列按顺序处理同组消息

2. 配置Lambda的SQS触发器 #

在第二个Lambda函数的触发器设置中添加SQS触发器，针对目标FIFO队列配置以下关键参数：

• 批量大小：设置单次触发Lambda时最多处理的消息数量（例如100，根据业务处理能力调整，最大值1000）
• 批量窗口：设置为5秒——这是核心参数，Lambda会在这个时间窗口内收集同一消息组的消息，合并为一个事件触发函数
• 可见性超时：设置为大于Lambda函数超时时间的值（例如如果Lambda超时是10秒，可见性超时设为15秒），避免消息在处理完成前被重新分发

3. 第一个Lambda发送消息的规则 #

第一个Lambda向FIFO队列发送消息时，必须为需要合并的消息指定相同的Message Group ID。只有同一组的消息才会被Lambda在批量窗口内合并处理；不同组的消息会各自触发独立的Lambda事件。

示例Python代码（第一个Lambda发送消息）：

import boto3

sqs = boto3.client('sqs')
QUEUE_URL = '你的FIFO队列URL'

def lambda_handler(event, context):
    # 模拟多条消息
    messages = ["消息1", "消息2", "消息3"]
    for msg in messages:
        sqs.send_message(
            QueueUrl=QUEUE_URL,
            MessageBody=msg,
            MessageGroupId="合并处理组ID"  # 同一组ID确保消息被批量合并
        )
    return {"statusCode": 200}

4. 第二个Lambda处理批量消息 #

第二个Lambda的事件参数会包含一个Records数组，里面是批量收集的所有消息，只需遍历数组处理即可。

示例Python代码（第二个Lambda处理消息）：

def lambda_handler(event, context):
    # 遍历批量消息
    for record in event['Records']:
        message_body = record['body']
        # 这里添加你的业务处理逻辑
        print(f"处理消息: {message_body}")
    return {"statusCode": 200, "processed_count": len(event['Records'])}

注意事项 #

• 批量窗口是针对每个Message Group ID生效的，不同组的消息不会跨组合并
• 如果在5秒窗口内没有达到设置的批量大小，Lambda也会触发，发送当前已收集的消息
• FIFO队列的消息顺序性会被保留，同一组内的消息会按发送顺序出现在Lambda的事件数组中

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