如何防止 Amazon SQS 消息多次调用我的 Lambda 函数？

简述

Lambda 支持至少一次消息传递。在特定情况下，重试机制可能会发送同一条消息的重复副本。如果您配置了死信队列，Amazon SQS 随后会将这些消息发送到该队列。

同一 Amazon SQS 消息触发重复 Lambda 调用的原因可能如下：

• 您的函数返回错误或超时。
• Lambda 服务成功处理了批次，但未能在可见性超时过期之前从 Amazon SQS 队列中删除该消息。
• Lambda 服务已将事件发送给函数，但未收到函数的确认。
• 间歇性问题导致 Amazon SQS 返回同一条消息，并且 Lambda 服务再次轮询了该消息。
• 批处理窗口与函数执行时长之和大于您的 Amazon SQS 队列可见性超时。SQS 可见性超时必须至少是函数超时与批处理窗口超时之和的六倍。

要确认您多次收到同一条消息，请检查消息 ID。如果您收到同一条消息的多个副本，那么这些消息具有相同的 ID。
注意： 如果您收到多条内容相同但消息 ID 不同的消息，则说明队列多次收到了该消息。请确认生产者正在运行、将消息发送到正确的队列 URL，并且具有所需的权限。

解决方法

要解决此问题，请在 Lambda 函数的 SQS 触发器中开启报告批处理项目失败。然后，创建一个模块化函数代码，遍历批次、处理并删除成功和重复的消息。该函数将成功消息的 messageID 存储在 Amazon DynamoDB 表中，然后验证该消息是否之前已处理过。

重要信息： 以下解决方法会针对每条传入消息向 DynamoDB 发送多个请求，从而降低处理速度，并因 API 调用而产生更高的成本。

创建 DynamoDB 表

以下 DynamoDB 表用于存储您的消息 ID，以便 Lambda 函数可以对其进行比较以检测重复。

完成以下步骤：

1. 打开 DynamoDB 控制台。
2. 选择创建表。
3. 在创建 DynamoDB 表页面上，输入以下值：
   对于表，输入 ProcessedRecords
   在分区键下，输入 Records
   将数据类型设置为字符串
4. 选择创建表。

创建 Lambda 函数

重要信息： Lambda 函数代码必须是幂等的。有关函数幂等逻辑的示例，请参阅使用 AWS Lambda 函数的最佳实践。

创建一个 Lambda 函数，将传入消息与之前成功处理并存储在 DynamoDB 表中的消息进行比较。如果某条消息之前已成功处理，则该函数不允许处理重复消息。如果新的、唯一的消息成功处理，则它们将被发送到表中以供将来比较。

SQS 会重试失败的消息，直到 Lambda 成功处理它们，或者直到消息的 ReceiveCount 超过 maxReceiveCount。

在您的函数中，添加执行角色以允许 dynamodb:Query 和 dynamodb:PutItem 操作。然后，在 SQS 触发器中报告批处理项目失败，以识别并跳过批次中的重复消息。

以下示例 Lambda 函数逻辑将消息正文转换为大写：

import boto3

dynamodb_client = boto3.client('dynamodb')

DDB_TABLE = 'ProcessedRecords'

# Validates if the message is already processed in previous invokes.
# @input string message_id
#
# @param message_id used to query the message from DynamoDB
# @return Boolean
def is_duplicate_message(message_id):
    return dynamodb_client.query(
        TableName = DDB_TABLE
        Select='COUNT',
        KeyConditionExpression='Records = :Records',
        ExpressionAttributeValues={
            ':Records': {'S': message_id}
        }
    )["Count"] != 0

# Processes the message body to upper case.
# @input string body
#
# @param body to be processed
# @return uppercase body
def process_message(body):
    return body.upper()

# Put the message to the DynamoDB Table.
# @input string batch_item_success
#
# @param batch_item_success of the message to put.
# @return Boolean
def push_to_dynamoDB(batch_item_success):

    for message_id in batch_item_success:
        response = dynamodb_client.put_item(
                        TableName = DDB_TABLE,
                        Item={ 'Records': {'S':message_id}
                        }
                    )
    return True

# processor function iterating through messages in the event.
# @input dict Records
#
# @param Records to be processed
# @return Boolean
def iterate_records(Records):

    batch_item_failures = []
    batch_item_success = []

    for record in Records:

        message_id = record["messageId"]
        print("Message: " + message_id)
        if is_duplicate_message(message_id):   
            print("Message duplicate: " + message_id)
            continue

        try:
            process_message(record["body"])
            batch_item_success.append(message_id)
        except:
            batch_item_failures.append({"itemIdentifier": message_id})

    push_to_dynamoDB(batch_item_success)
    return batch_item_failures

def lambda_handler(event, context):

    return {"batchItemFailures": iterate_records(event["Records"])}

相关信息

为什么我的 Lambda 函数会重试有效的 Amazon SQS 消息并将其放入死信队列？

为什么我的 Amazon SQS 队列没有调用我的 Lambda 函数？

为什么我的 Amazon SQS 消息意外发送到 DLQ？