如何使用 lambda 函数（不是通过预定事件）处理 SQS 队列？答案

【问题标题】：How to process SQS queue with lambda function (not via scheduled events)?如何使用 lambda 函数（不是通过预定事件）处理 SQS 队列？
【发布时间】：2021-03-26 10:37:47
【问题描述】：

这是我正在尝试制作的简化方案：

http 请求 --> (Gateway API + lambda A) --> SQS --> (lambda B ?????) --> DynamoDB

所以它应该如下所示工作：来自许多 http 请求的数据（例如，每秒最多 500 个）由我的 lambda 函数 A 放入 SQS 队列。然后另一个函数 B 处理队列：最多读取 10 个项目（定期）并使用 BatchWriteItem 将它们写入 DynamoDB。

问题是我不知道如何触发第二个 lambda 函数。它应该被频繁地调用，每秒多次（或至少每秒一次），因为我需要队列中的所有数据尽快进入 DynamoDB（这就是为什么通过预定事件调用 lambda 函数 B 描述 here 不是一个选项）

我为什么不想直接写入 DynamoDB，而不使用 SQS？

完全避免使用 SQS 对我来说很好。我试图用 SQS 解决的问题是 DynamoDB 节流。甚至不限制自身，而是使用 AWS SDK 将数据写入 DynamoDB 时的处理方式：当一条一条写入记录并限制它们时，AWS SDK 会静默地重试写入，从而导致从 http 客户端的请求处理时间增加查看。

所以我想暂时将数据存储在队列中，将响应“200 OK”发送回客户端，然后通过单独的函数处理队列，使用一个 DynamoDB 的 BatchWriteItem 调用写入多条记录（其中返回未处理的项目，而不是在节流的情况下自动重试）。 我什至宁愿丢失一些记录，而不是增加接收记录和存储在 DynamoDB 中的记录之间的延迟

UPD：如果有人感兴趣，我已经找到了如何让 aws-sdk 在节流时跳过自动重试：有一个特殊参数 maxRetries。无论如何，按照下面的建议使用 Kinesis

【问题讨论】：

什么原因无法解决 - 依靠 Number 计划的事件源？
也许我遗漏了一些东西，但从我在docs 中看到的最小调度速率为 5 分钟。但我需要能够每秒运行我的函数。我对所有这些东西完全陌生，所以请指教。
了解 5 分钟的延迟。如果是这种情况，那么您必须使用在 EC2 中运行的自定义进程/脚本，它将 SQS 中的项目连接到 DynamoDB。我也在考虑 AWS Data Pipeline，它的 schedule 选项接受值为 1 的文本框和 min 的组合作为时间频率。
是的，EC2 中的自定义进程而不是 lambda 函数将解决所有问题，这将是我最后的后备选项。
我之前也尝试过在 Lambda 中完成这类事情。我放弃。 SQS 与 Lambda 的任何集成目前都感觉像是一个巨大的黑客攻击。我现在只有一个在 t2.nano 实例上运行的服务，它不断地轮询 SQS。希望亚马逊将来会在 SQS 和 Lambda 之间添加某种形式的集成。如果您不反对在 AWS 之外寻找，您可以查看 IronIO，它具有 IronMQ/IronWorker 集成，看起来与我希望 AWS 的集成类似。

标签： amazon-web-services aws-lambda amazon-sqs

【解决方案1】：

[这不能直接回答你的明确问题，所以根据我的经验，它会被否决:) 但是，我会回答你试图解决的基本问题。]

我们接收大量传入请求并将它们提供给 AWS Lambda 函数以便以有节奏的方式写入 DynamoDB 的方法是用 Amazon Kinesis 流替换提议架构中的 SQS。

Kinesis 流可以驱动 AWS Lambda 函数。

Kinesis 流保证任何给定键的传递消息的顺序（非常适合有序的数据库操作）。

Kinesis 流让您可以指定可以并行运行多少个 AWS Lambda 函数（每个分区一个），这可以与您的 DynamoDB 写入容量相协调。

Kinesis 流可以在一次 AWS Lambda 函数调用中传递多条可用消息，从而实现进一步优化。

注意：实际上是 AWS Lambda 服务从 Amazon Kinesis 流中读取然后调用该函数，而不是 Kinesis 流直接调用 AWS Lambda；但有时更容易将其想象为 Kinesis 驱动它。对用户的结果几乎相同。

【讨论】：

这很棒。在从 Spark 集群到 S3 再到 Lambda 的 SNS 节流方面，我一直遇到很大的问题（因为 AWS 将我们的 Lambda 函数从 100 增加到 6,000！）并且一直在做各种各样的事情来限制我们传递消息的速率。您将 Lambda 与 Kinesis 结合使用的解决方案似乎正是我想要的。
嗨，Eric，我喜欢这种方法，我只有一个问题：是否可以分时间批处理消息？说 1 小时，然后将批次传递给 Lambda？
@luisfarzati 将消息放入 Kinesis 流中，它们将在其中缓冲。安排一个 AWS Lambda 函数每小时运行一次。它从蒸汽中读取消息并处理它们。并不是说 AWS Lambda 函数需要在最多 5 分钟内完成当前的限制。
问题是 Kinesis 流对 SQS 拥有的每条消息都没有“至少一次”的高可用性保证。例如，在处理大数据时，在消防站丢失一些消息是可以接受的，但在某些应用程序中丢失单个请求是不能接受的。
AWS 定价总是在变化（向下），但我对上面的成本 cmets 感到困惑。对于所讨论的大多数用途，Amazon Kinesis 应该明显比 Amazon SQS 便宜。

【解决方案2】：

很遗憾，您无法直接集成 SQS 和 Lambda。但不要太担心。有解决办法！您需要添加另一个亚马逊服务，您的所有问题都会得到解决。

http requests --> (Gateway API + lambda A) --> SQS + SNS --> lambda B --> DynamoDB

您可以向第二个 lambda 服务触发 SNS 通知以启动它。一旦启动，它就可以排空队列并将所有结果写入 DynamoDB。要更好地了解 Lambda 的可能事件源，请查看 these docs。

【讨论】：

感谢您的建议！我曾考虑使用 SNS 调用第二个 lambda 函数，但对每个传入的 http 请求（在我的情况下约为每秒 500 个）都会调用 lambda B 的事实感到困惑，尽管我希望理想情况下调用一次一秒。我担心这种开销：每秒 500 次与实际需要的一次。所以我将按照@Eric Hammond 的建议研究 Kinesis
我总是忘记 Kinesis。我们将 Kafka 用于我们的事件流（Kinesis 是基于它的）。这比在您的情况下使用 SNS 更好！
您不能将 sns 订阅到 sqs。
我没说可以。我建议 lambda 在将消息放入 SQS 的同时触发 SNS。由于问题是如何使用 lambda 处理 SQS 中的事件，我认为我不必重申这一点。
这是一种享受 - 我遇到了无法使用 FIFO SQS 直接订阅 SNS 的问题，目前不支持。在添加到 SQS 队列的回调中，我现在发布到 SNS 主题并让 2 lambda 订阅该 SNS 主题。太好了，谢谢！

【解决方案3】：

自 2018 年 6 月 28 日起，您现在可以使用 SQS 本地触发 AWS Lambda 函数。不再需要解决方法！

https://aws.amazon.com/blogs/aws/aws-lambda-adds-amazon-simple-queue-service-to-supported-event-sources/

并且在 2019 年 11 月，添加了对 FIFO 队列的支持：

https://aws.amazon.com/blogs/compute/new-for-aws-lambda-sqs-fifo-as-an-event-source/

【讨论】：

目前 Lambda 触发器仅适用于标准队列，不适用于 FIFO 队列。
Brett，您的评论是我在官方文档之前来到 Stack Overflow 的原因。（就像 this aws doc 没有提到它。）是否有任何 SO 机制可以加快对顶部的新真实答案？

【解决方案4】：

另一种解决方案是将项目添加到 SQS，使用 Event 调用目标 Lambda 函数，因此它是异步的。

然后，异步 Lambda 可以从 SQS 获取任意数量的项目并对其进行处理。

我还将向异步 Lambda 添加计划调用，以处理队列中出错的任何项目。

[更新] 您现在可以在队列上的新消息上设置 Lambda 触发器

【讨论】：

我喜欢这个解决方案。在我的情况下，数据库可能已关闭，因此按计划运行 lambda b 有利于重试处理队列。我没有想到的是，如果队列开始变得繁忙，你将如何扩展到多个 lambda b 消费者？
您可以在 Cloudwatch 上设置触发 SNS 的警报，该 SNS 触发拉动 SQS 的 Lambda，这样您可以增加轮询器的数量。您还可以创建一个特定的 Lambda，该 Lambda 将根据可用项目的数量异步启动多个 Lambda
对任何这样做的人的问题：您是否发现 CloudWatch 事件按时交付？使用 rate(5 mins) 或 cron(5/* etc)，我观察到每 5 分钟窗口有 0 到 3 条消息。不理想！
自此问题以来已添加 SQS Lambda 触发器

【解决方案5】：

也许一个更具成本效益的解决方案是将所有内容保留在 SQS 中（原样），然后运行一个调度事件，该事件调用一个处理队列中的项目的多线程 Lambda 函数？

这样，您的队列工作者可以完全匹配您的限制。如果队列为空，函数可以提前结束或开始单线程轮询。

在这种情况下，Kinesis 听起来有点过头了——例如，您不需要原始订单。另外，同时运行多个 Lambda 肯定比只运行一个多线程 Lambda 更昂贵。

您的 Lambda 将完全与 I/O 相关，对 AWS 服务进行外部调用，因此一个函数可能非常适合。

【讨论】：

【解决方案6】：

以下是我从 SQS 队列中收集消息的方式：

package au.com.redbarn.aws.lambda2lambda_via_sqs;

import java.util.List;

import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.SQSEvent;
import com.amazonaws.services.lambda.runtime.events.SQSEvent.SQSMessage;

import lombok.extern.log4j.Log4j2;

@Log4j2
public class SQSConsumerLambda implements RequestHandler<SQSEvent, String> {

    @Override
    public String handleRequest(SQSEvent input, Context context) {

        log.info("message received");

        List<SQSMessage> records = input.getRecords();

        for (SQSMessage record : records) {
            log.info(record.getBody());
        }

        return "Ok";
    }
}

将您的 DynamoDB 代码添加到 handleRequest()，Lambda B 就完成了。

【讨论】：

【解决方案7】：

这是我对这个问题的解决方案：

HTTP request --> DynamoDb --> Stream --> Lambda Function

在此解决方案中，您必须为表格设置流。流由您将编写的 Lambda 函数处理，仅此而已。无需使用 SQS 或其他任何东西。

当然，这是一个简化的设计，它只适用于简单的问题。对于更复杂的场景，请使用 Kinesis（如其他答案中所述）。

这是link to AWS documentation on the topic。

【讨论】：

【解决方案8】：

我相信 AWS 现在已经想出了一种 SQS 可以触发 lambda 函数的方法。所以我想我们可以使用 SQS 来平滑数据的突发负载到发电机，以防你不关心消息的顺序。查看他们的博客以了解此新更新：https://aws.amazon.com/blogs/aws/aws-lambda-adds-amazon-simple-queue-service-to-supported-event-sources/

【讨论】：