使用 AWS 无服务器和 NodeJS 在接收器 lambda 处未接收到来自 SQS 的所有消息

Posted 2023-03-07

【中文标题】使用 AWS 无服务器和 NodeJS 在接收器 lambda 处未接收到来自 SQS 的所有消息

【英文标题】：Not receiving all messages at receiver lambda from SQS with AWS serverless and NodeJS

【发布时间】：2020-01-22 03:51:48

【问题描述】：

我们使用 AWS Serverless 和 NodeJS 创建了两个 lamdba 函数，一个是发送者，另一个是接收者，在这两者之间有一个标准 SQS，它从发送者 lambda 接收数据，如果有任何消息到达，它会自动触发接收者 lambda 函数在 SQS。

在这里，我们能够成功地将所有消息从发送者 lambda 发送到 SQS（根据日志），但在接收者 lambda 处，我们没有收到所有消息。

为了测试这个场景，我从邮递员那里发送了 1000 条消息，并测试了发送者将所有消息完美地发送到 SQS，但接收者只从中获得了 986 条随机消息，其中一些随机消息被遗漏了。

我也尝试过将 SQS 类型改为 FIFO 而不是标准，但 lambda 不支持。

这里是发送者和接收者函数的代码：

发件人：

sender: () => 
    const sqs = new AWS.SQS( apiVersion: "2019-08-09" );
    let body = [
         id : "1" ,
         id : "2" ,
         id : "3" 
    ]
    let params = 
        DelaySeconds: 10,
        QueueUrl: url,
        MessageBody: JSON.stringify(body)
    ;
    sqs.sendMessage(params, function (err, data) 
        if (err) 
            callback(true, null);
         else 
            callback(false, data);
        
    );

接收者：

receiver: () => 
    event.Records[0].body = JSON.parse(event.Records[0].body);
    async.timesSeries(event.Records[0].body.length, (i, next) => 
        const params = 
            TableName: "user",
            Key: 
                id: "1"
            
        ;
        dynamoDb.get(params).promise()
            .then(result => 
                //save user
            )
            .catch(error => 
                //throw err
                next();
            );
    , () => 
        console.log("deleting message");
        const deleteParams = 
            QueueUrl: "queue_url",
            ReceiptHandle: event.Records[0].receiptHandle
        ;
        sqs.deleteMessage(deleteParams, function (err, data) 
            if (err) 
                //throw error
             else 
                //success
            
        );
    );

【问题讨论】：

你找到解决办法了吗？

【参考方案1】：

在这段代码中：

event.Records[0].body = JSON.parse(event.Records[0].body);
async.timesSeries(event.Records[0].body.length, (i, next) => 

通过显式采用Records[0]，您只处理列表中的第一条记录。 event.Records 是一个列表，需要遍历列表中的所有记录。请参阅文档here 中包含多条记录的示例 SQS 事件。

---

或者，在您的 Lambda SQS 集成设置中，将 Batch Size 更改为 1，以便一次仅将 1 记录传递给您的 Lambda 函数。

【讨论】：

嗨，马克，感谢您的快速回复。我检查了它并在接收器的第一行添加了控制台，发现它甚至没有打印该控制台以获得少量消息。接收者：() => console.log("第一行"); event.Records[0].body = JSON.parse(event.Records[0].body);

@Anshul 我不知道如何调用 Lambda 函数但不知何故不执行第一行代码。你怎么知道它没有打印“一些消息”？你的Batch Size 设置是什么？此外，您不应该删除代码中的消息，您应该返回一个成功响应，这将触发 Lambda SQS 集成为您删除消息。