本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。
使用消息处理框架消费 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消息轮询器
可以在呼叫SQSMessagePollerOptions
AddSQSPoller
时配置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