AWS SNS 中定期丢失消息

Question

我知道，这听起来很奇怪，但我的 AWS SNS 有问题 =)

我有 lambda 函数，它正在向 AWS SNS 发送消息。我也有几个 SQS 作为我的 SNS 的订阅。另外，我有 SNS 和 SQS 的死队列。并为 SNS（交付和错误）开启日志记录 (100%)。

在大多数情况下，我的架构按预期工作 - Lambda 正在向 SNS 发送消息

1. 我在 Lambda 日志中看到来自 SNS 的成功响应（boto3 / sns 客户端）
2. 我在 SNS 日志中看到成功登录
3. 我可以在 SQS 中收到我的消息

但有时 Lambda 和 SNS 之间会出现问题，因为：

1. 我在 Lambda 中看到成功响应，类似于：

    {'MessageId': '292af724-XXXc49658c0', 'SequenceNumber': '10000000000000000551',
    'ResponseMetadata': {'RequestId': 'ba126582-XXX8f2', 
    'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': 'ba126582-XXX18f2', 
    'content-type': 'text/xml', 'content-length': '352', 
    'date': 'Thu, 29 Apr 2021 13:00:28 GMT'}, 'RetryAttempts': 0}}

1. 这就是我所拥有的一切 :( SNS 没有任何错误（SNS 权限没问题，因为我之前看到失败错误和成功传递消息）。在 DLQ SNS/SQS 中没有任何消息。什么都没有 :(

所以，我的问题是 - 这怎么可能？我该如何解决？

备注 - 我正在使用 FIFO SNS / SQS

Answer 1

您是否可能只是看到重复数据删除的结果？ https://docs.aws.amazon.com/sns/latest/dg/fifo-message-dedup.html

如果您使用相同的重复数据删除 ID，或者如果您打开了内容重复数据删除功能，那么您将无法在 5 分钟内传递相同的消息。

SNS/SQS 具有如此持久的持久性，除非您每小时处理数十亿条消息，否则几乎不可能随机丢失消息。

Answer 2

检查您的 Lambda 上的 SQS 触发器的批量大小是否为 1。如果您的输出 Lambda 设计为一次仅处理 1 个请求，则队列中的多个项目可以在一个组中弹出，并产生错觉的迷失。如果您的 Lambda 又短又快，这可能是可取的……您只需要注意它会发生。

我的设置与您的设置略有不同，但我认为无论如何都值得分享：

1. 批量 SNS 触发 Lambda。
2. 用于跟踪待处理作业的 SQS。
3. SQS 触发器。
4. 正在使用 Lambda。

因此，从技术上讲，您可以通过更改第 3 步或第 4 步来解决问题，具体取决于适合您的方法。更改批量大小的最简单方法是创建一个新触发器。虽然可以通过 CLI 完成，但我不知道命令。

Answer 3

我知道这个问题有点老了，但我想回答一下，以防你或其他人面临这个问题并且感到困惑。

Anarki 和 DazedAndConfused 的回答都是有效的。您应该查看的一个角度是在您的代码中。如果您的流程是：

Lambda 1 > SNS > SQS > Lambda 2

您的 Lambda 2 函数必须循环处理 SQS 记录。即使 Lambda 1 发布了多个 SNS 事件，也可以将一些消息进行批处理，并在 Lambda 2 中进行处理。

话虽如此，您的 Lambda 2 循环中很可能有一个 return 语句。在这种情况下，处理 Record[0] 后，您的 Lambda 2 将终止，有效地跳过所有后续记录。

以 Python 为例

例如，这个：

def lambda_handler(event, context):
    print('Event data is: ' + str(event))


    for record in event['Records']:
        print(record['messageId'])
    
        return {
            "statusCode": 200,
            "body": "Success!"
        }

...与此大不相同：

def lambda_handler(event, context):
    print('Event data is: ' + str(event))


    for record in event['Records']:
        print(record['messageId'])

    return {
        "statusCode": 200,
        "body": "Success!"
    }

我和你有同样的困惑和疑问，不幸的是，我知道并学会了艰难的道路。