AWS.Lambda.Powertools.BatchProcessing 1.1.1

AWS.Lambda.Powertools.BatchProcessing

批量处理实用工具在处理来自 Amazon SQS、Amazon Kinesis Data Streams 和 Amazon DynamoDB Streams 的批量时处理部分失败。

主要功能

• 报告批量项失败以减少出错时记录的重试次数
• 简单的接口以处理每个批量记录
• 自带批量处理器
• 并行处理

后台

当使用 SQS、Kinesis Data Streams 或 DynamoDB Streams 作为 Lambda 事件源时，您的 Lambda 函数会触发一个消息批次。

如果您的函数无法处理批量中的任何消息，则整个批次将返回到您的队列或流。该批次然后会重试，直到以下情况之一发生：a) 您的 Lambda 函数返回成功的响应，b) 记录达到最大重试次数，或 c) 记录过期。

当您在 Lambda 函数事件源配置中启用“报告批量项失败”功能时，此行为会发生变化。

• SQS队列。只有标记为失败的消息会返回队列进行重试，而成功消息将被删除。
• Kinesis数据流和DynamoDB流。单次报告的失败将使用其序列号作为流检查点。多次报告的失败将使用最低序列号作为检查点。

阅读文档

要获取完整功能列表，请访问docs.powertools.aws.dev/lambda/dotnet/utilities/batch-processing/

GitHub: https://github.com/aws-powertools/powertools-lambda-dotnet/

示例函数

在此处查看完整示例：github.com/aws-powertools/powertools-lambda-dotnet/tree/develop/examples/BatchProcessing

[BatchProcessor(RecordHandler = typeof(CustomSqsRecordHandler))]
public BatchItemFailuresResponse HandlerUsingAttribute(SQSEvent _)
{
    return SqsBatchProcessor.Result.BatchItemFailuresResponse;
}

public class CustomSqsRecordHandler : ISqsRecordHandler
{
    public async Task<RecordHandlerResult> HandleAsync(SQSEvent.SQSMessage record, CancellationToken cancellationToken)
    {
        /*
            Your business logic. 
            If an exception is thrown, the item will be marked as a partial batch item failure.
        */
        
        var product = JsonSerializer.Deserialize<JsonElement>(record.Body);

        if (product.GetProperty("Id").GetInt16() == 4)
        {
            throw new ArgumentException("Error on 4");
        }

        return await Task.FromResult(RecordHandlerResult.None);
    }
}