使用消息处理框架消费 Amazon 消息,用于. NET - Amazon SDK for .NET
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅 中国的 Amazon Web Services 服务入门 (PDF)

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

使用消息处理框架消费 Amazon 消息,用于. NET

这是适用于预览版中功能的预发布文档。本文档随时可能更改。

的 Amazon 消息处理框架. NET允许您使用使用框架或其中一个消息传递服务发布的消息。可以通过多种方式使用消息,其中一些方式如下所述。

消息处理器

要使用消息,请使用要处理的每种消息类型的IMessageHandler接口实现消息处理程序。消息类型和消息处理程序之间的映射是在项目启动时配置的。

await Host.CreateDefaultBuilder(args) .ConfigureServices(services => { // Register the AWS Message Processing Framework for .NET services.AddAWSMessageBus(builder => { // Register an SQS Queue that the framework will poll for messages. // NOTE: The URL given below is an example. Use the appropriate URL for your SQS Queue. builder.AddSQSPoller("https://sqs.us-west-2.amazonaws.com/012345678910/MyAppProd"); // Register all IMessageHandler implementations with the message type they should process. // Here messages that match our ChatMessage .NET type will be handled by our ChatMessageHandler builder.AddMessageHandler<ChatMessageHandler, ChatMessage>(); }); }) .Build() .RunAsync();

以下代码显示了消息的示例消息处理程序。ChatMessage

public class ChatMessageHandler : IMessageHandler<ChatMessage> { public Task<MessageProcessStatus> HandleAsync(MessageEnvelope<ChatMessage> messageEnvelope, CancellationToken token = default) { // Add business and validation logic here. if (messageEnvelope == null) { return Task.FromResult(MessageProcessStatus.Failed()); } if (messageEnvelope.Message == null) { return Task.FromResult(MessageProcessStatus.Failed()); } ChatMessage message = messageEnvelope.Message; Console.WriteLine($"Message Description: {message.MessageDescription}"); // Return success so the framework will delete the message from the queue. return Task.FromResult(MessageProcessStatus.Success()); } }

外部MessageEnvelope包含框架使用的元数据。它的message属性是消息类型(在本例中ChatMessage)。

您可以返回MessageProcessStatus.Success()以表明消息已成功处理,框架将从 Amazon SQS 队列中删除该消息。返回时MessageProcessStatus.Failed(),消息将保留在队列中,可以在该队列中再次处理或移至死信队列(如果已配置)。

在长时间运行的进程中处理消息

您可以使用队列调AddSQSPoller用URL以启动一个长时间运行的SQS队列 BackgroundService,该队列将持续轮询队列并处理消息。

await Host.CreateDefaultBuilder(args) .ConfigureServices(services => { // Register the AWS Message Processing Framework for .NET services.AddAWSMessageBus(builder => { // Register an SQS Queue that the framework will poll for messages. // NOTE: The URL given below is an example. Use the appropriate URL for your SQS Queue. builder.AddSQSPoller("https://sqs.us-west-2.amazonaws.com/012345678910/MyAppProd", options => { // The maximum number of messages from this queue that the framework will process concurrently on this client. options.MaxNumberOfConcurrentMessages = 10; // The duration each call to SQS will wait for new messages. options.WaitTimeSeconds = 20; }); // Register all IMessageHandler implementations with the message type they should process. builder.AddMessageHandler<ChatMessageHandler, ChatMessage>(); }); }) .Build() .RunAsync();

配置SQS消息轮询器

可以在呼叫SQSMessagePollerOptionsAddSQSPoller时配置SQS消息轮询器。

  • MaxNumberOfConcurrentMessages-队列中可同时处理的最大消息数。默认值是 10。

  • WaitTimeSeconds-ReceiveMessage SQS 呼叫等待消息到达队列后再返回的持续时间(以秒为单位)。如果有消息可用,则该呼叫的返回时间早于WaitTimeSeconds。默认值为 20。

消息可见性超时处理

SQS消息有可见性超时时间。当一个使用者开始处理给定消息时,它会保留在队列中,但为了避免多次处理该消息,其他消费者会将其隐藏。如果消息在再次可见之前未被处理和删除,则其他使用者可能会尝试处理同一条消息。

该框架将跟踪并尝试延长其当前正在处理的消息的可见性超时时间。您可以在 “呼叫SQSMessagePollerOptions时” 上配置此行为AddSQSPoller

  • VisibilityTimeout-在后续检索请求中隐藏收到消息的持续时间(以秒为单位)。默认值为 30。

  • VisibilityTimeoutExtensionThreshold-当消息的可见性超时在过期后的几秒钟内时,框架将延长可见性超时(再延长VisibilityTimeout几秒钟)。默认值是 5。

  • VisibilityTimeoutExtensionHeartbeatInterval-框架检查即将到期的几秒钟之内的消息,然后延长其可见性超时的频率(以VisibilityTimeoutExtensionThreshold秒为单位)。默认值是 1。

在以下示例中,框架将每 1 秒检查一次仍在处理的消息。对于这些消息在再次可见后 5 秒钟内,框架会自动将每条消息的可见性超时时间再延长 30 秒。

// NOTE: The URL given below is an example. Use the appropriate URL for your SQS Queue. builder.AddSQSPoller("https://sqs.us-west-2.amazonaws.com/012345678910/MyAppProd", options => { options.VisibilityTimeout = 30; options.VisibilityTimeoutExtensionThreshold = 5; VisibilityTimeoutExtensionHeartbeatInterval = 1; });

处理 Amazon Lambda 函数中的消息

您可以将 Amazon 消息处理框架用于。 NET与 SQS Lambda 集成。这是由AWS.Messaging.Lambda软件包提供的。请参考它README以开始使用。