批量处理AWS Lambda消息 [英] Processing AWS Lambda messages in Batches

问题描述

我在想什么，我真的找不到关于它的信息.也许这不是要走的路，但是，我只想知道.

I am wondering something, and I really can't find information about it. Maybe it is not the way to go but, I would just like to know.

这是关于Lambda批量工作的.我知道我可以设置Lambda来使用批处理消息.在我的Lambda函数中，我迭代每条消息，如果一条消息失败，则Lambda退出.然后循环再次开始.

It is about Lambda working in batches. I know I can set up Lambda to consume batch messages. In my Lambda function I iterate each message, and if one fails, Lambda exits. And the cycle starts again.

我想知道略有不同的方法 假设我有3条消息: A ， B 和 C .我也分批服用.现在，如果消息B失败(例如API调用失败)，我将消息B返回给SQS，并继续处理消息C.

I am wondering about slightly different approach Let's assume I have three messages: A, B and C. I also take them in batches. Now if the message B fails (e.g. API call failed), I return message B to SQS and keep processing the message C.

有可能吗?如果是的话，这是一个好方法吗?因为我看到我需要在Lambda中实现一些额外的复杂性，否则不需要实现.

Is it possible? If it is, is it a good approach? Because I see that I need to implement some extra complexity in Lambda and what not.

谢谢

推荐答案

有一篇很棒的文章

There's an excellent article here. The relevant parts for you are...

• 使用batchSize为1，以便消息自行成功或失败.
• 确保您的处理是幂等的，因此重新处理邮件不会造成危害，而不会产生额外的处理费用.
• 可以通过捕获功能代码中的错误并将消息发送到死信队列中以进行进一步处理来处理错误.
• 成功处理消息后，在函数中手动调用DeleteMessage API.

• Using a batchSize of 1, so that messages succeed or fail on their own.
• Making sure your processing is idempotent, so reprocessing a message isn't harmful, outside of the extra processing cost.
• Handle errors within your function code, perhaps by catching them and sending the message to a dead letter queue for further processing.
• Calling the DeleteMessage API manually within your function after successfully processing a message.

最后一个要点是我如何处理相同的问题.与其立即返回错误，不如存储它们或注意已发生错误，而是继续处理批处理中的其余消息，而不是立即返回错误.在处理结束时，返回或引发错误，以使SQS-> lambda触发器知道不删除失败的消息.您的lambda处理程序将删除所有成功的消息.

The last bullet point is how I've managed to deal with the same problem. Instead of returning errors immediately, store them or note that an error has occurred, but then continue to handle the rest of the messages in the batch. At the end of processing, return or raise an error so that the SQS -> lambda trigger knows not to delete the failed messages. All successful messages will have already been deleted by your lambda handler.

sqs = boto3.client('sqs')

def handler(event, context):
    failed = False

    for msg in event['Records']:
        try:
            # Do something with the message.
            handle_message(msg)
        except Exception:
            # Ok it failed, but allow the loop to finish.
            logger.exception('Failed to handle message')
            failed = True
        else:
            # The message was handled successfully. We can delete it now.
            sqs.delete_message(
                QueueUrl=<queue_url>,
                ReceiptHandle=msg['receiptHandle'],
            )

    # It doesn't matter what the error is. You just want to raise here
    # to ensure the trigger doesn't delete any of the failed messages.
    if failed:
        raise RuntimeError('Failed to process one or more messages')

def handle_msg(msg):
    ...

这篇关于批量处理AWS Lambda消息的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！