AWS SDK for .NET 操作 Amazon SQS 实战指南：从单操作示例到消息队列完整场景

示例工程教程后端【免费下载链接】aws-doc-sdk-examplesWelcome to the AWS Code Examples Repository. This repo contains code examples used in the AWS documentation, AWS SDK Developer Guides, and more. For more information, see the Readme.md file below.项目地址https://gitcode.com/gh_mirrors/aw/aws-doc-sdk-examples点击查看免费下载导读本文以 dotnetv3/SQS/README.md 为核心系统讲解如何使用 AWS SDK for .NET 3.x 开发 Amazon Simple Queue ServiceSQS应用涵盖ListQueues入门示例、创建/发送/接收/删除队列等全部单操作代码以及SNS 主题 SQS 队列消息发布场景和 AWS Message Processing Framework for .NET 框架化开发两大完整场景。读完本文你将掌握 SQS 在 .NET 中的全部核心 API 调用方式、队列策略与 FIFO 配置要点并能直接运行仓库中的可执行示例与测试。Amazon SQS 是一项全托管的消息队列服务用于解耦和扩展微服务、分布式系统与无服务器应用。仓库中的示例覆盖了从Hello World级别的入门调用到跨服务消息工作流的完整路径是学习 .NET 集成 SQS 最直接的参考实现。⚠️ 重要提示运行这些代码可能产生 AWS 账户费用详见 AWS 定价与免费套餐说明。运行测试同样可能产生费用。建议按最小权限原则Grant least privilege授权仅授予任务所需的最低权限。本仓库代码未经所有 AWS 区域验证部分服务仅在特定区域可用。一、示例概览与代码布局dotnetv3目录下的 SQS 示例分为两类单操作代码片段Single actions与完整场景Scenarios此外还有一个用于快速上手的入门示例Get started。类别内容位置入门Hello SQS列出队列ListQueuesHelloSQS.cs单操作CreateQueue、GetQueueAttributes、SetQueueAttributes、ReceiveMessage、DeleteMessageBatch、DeleteQueueSQSWrapper.cs单操作SendMessage、CreateQueue独立示例CreateSendExample.cs单操作GetQueueUrlGetQueueUrl.cs单操作ReceiveMessageDeleteMessageReceiveDeleteExample.cs场景SNS 主题发布消息到 SQS 队列TopicsAndQueues.cs场景AWS Message Processing Framework for .NETMessageProcessingFramework上述代码同时被 TopicsAndQueues 跨服务示例 复用其中SQSWrapper.cs是 SQS 操作的核心封装类单操作代码片段即取自于此。二、环境准备Prerequisites运行本目录示例前需要先完成 dotnetv3/README.md 中列出的前置条件核心包括安装合适的 .NET SDK多数示例使用 .NET 6部分要求 .NET 5。引入 AWS SDK for .NET 及对应服务包SQS 需要Amazon.SQS命名空间。配置 AWS 凭证可通过本地 credentials 文件配置或设置AWS_ACCESS_KEY_ID与AWS_SECRET_ACCESS_KEY环境变量。如果使用 AWS IAM Identity Center 进行身份认证还需为项目额外添加AWSSDK.SSO和AWSSDK.SSOOIDCNuGet 包见 MessageProcessingFramework 说明。三、入门示例Hello Amazon SQSListQueues仓库提供了一个最小入门示例演示如何使用异步 API 列出当前账户下的队列var sqsClient new AmazonSQSClient(); Console.WriteLine(Hello Amazon SQS! Following are some of your queues:); // Lets get the first five queues. var response await sqsClient.ListQueuesAsync( new ListQueuesRequest() { MaxResults 5 }); foreach (var queue in response.QueueUrls) { Console.WriteLine($\tQueue Url: {queue}); }关键点对应 HelloSQS.cs直接构造AmazonSQSClient不传区域时使用默认凭证链的默认区域。使用await调用异步方法ListQueuesAsync。通过MaxResults 5限制返回队列数量响应中的QueueUrls保存各队列的 URL。这是验证 SDK 安装、凭证配置与网络连通性最快的方式。四、单操作详解Single Actions以下示例展示如何调用 SQS 的单个服务函数。所有代码均可独立运行路径与行号见文末代码地图。4.1 创建队列CreateQueueSQSWrapper.cs 中的CreateQueueWithName展示了标准队列与 FIFO 队列的创建差异public async Taskstring CreateQueueWithName(string queueName, bool useFifoQueue) { int maxMessage 256 * 1024; var queueAttributes new Dictionarystring, string { { QueueAttributeName.MaximumMessageSize, maxMessage.ToString() } }; var createQueueRequest new CreateQueueRequest() { QueueName queueName, Attributes queueAttributes }; if (useFifoQueue) { // Update the name if it is not correct for a FIFO queue. if (!queueName.EndsWith(.fifo)) { createQueueRequest.QueueName queueName .fifo; } // Add an attribute for a FIFO queue. createQueueRequest.Attributes.Add( QueueAttributeName.FifoQueue, true); } var createResponse await _amazonSQSClient.CreateQueueAsync( new CreateQueueRequest() { QueueName queueName }); return createResponse.QueueUrl; }要点最大消息大小示例将MaximumMessageSize设为256 * 1024256 KB这也是 SQS 支持的上限。FIFO 队列启用QueueAttributeName.FifoQueue true属性且队列名必须以.fifo结尾否则调用会被拒绝——代码中通过EndsWith(.fifo)检查并自动补全后缀。创建成功返回队列 URLQueueUrl后续的收发消息、设置策略等操作都以该 URL 为定位依据。独立的 CreateSendExample.cs 则展示了另一种配置方式通过DelaySeconds延迟投递 60 秒和MessageRetentionPeriod消息保留 86400 秒即 1 天设置队列属性var request new CreateQueueRequest { QueueName queueName, Attributes new Dictionarystring, string { { DelaySeconds, 60 }, { MessageRetentionPeriod, 86400 }, }, }; var response await client.CreateQueueAsync(request);4.2 获取队列属性GetQueueAttributes创建队列后通常需要获取其 ARNAmazon Resource Name用于配置策略或订阅。参见 SQSWrapper.cspublic async Taskstring GetQueueArnByUrl(string queueUrl) { var getAttributesRequest new GetQueueAttributesRequest() { QueueUrl queueUrl, AttributeNames new Liststring() { QueueAttributeName.QueueArn } }; var getAttributesResponse await _amazonSQSClient.GetQueueAttributesAsync(getAttributesRequest); return getAttributesResponse.QueueARN; }通过AttributeNames指定要查询的属性此处为QueueArn响应中直接返回QueueARN。4.3 设置队列策略SetQueueAttributes要让 SNS 主题能够向队列投递消息必须为队列设置允许sqs:SendMessage的访问策略。参见 SQSWrapper.csvar queuePolicy { \Version\: \2012-10-17\, \Statement\: [{ \Effect\: \Allow\, \Principal\: { $\Service\: \sns.amazonaws.com\ }, \Action\: \sqs:SendMessage\, $\Resource\: \{queueArn}\, \Condition\: { \ArnEquals\: { $\aws:SourceArn\: \{topicArn}\ } } }] }; var attributesResponse await _amazonSQSClient.SetQueueAttributesAsync( new SetQueueAttributesRequest() { QueueUrl queueUrl, Attributes new Dictionarystring, string() { { Policy, queuePolicy } } }); return attributesResponse.HttpStatusCode HttpStatusCode.OK;策略要点Principal指定服务主体sns.amazonaws.com仅允许 SNS 服务访问。Action限制为sqs:SendMessage遵循最小权限原则。Condition使用ArnEquals的aws:SourceArn约束来源主题 ARN防止其他主题滥用该队列——这是防御跨账户/跨资源混淆代理攻击confused deputy的标准写法。4.4 发送消息SendMessageCreateSendExample.cs 展示了带消息属性的发送方式var sendMessageRequest new SendMessageRequest { DelaySeconds 10, MessageAttributes messageAttributes, MessageBody messageBody, QueueUrl queueUrl, }; var response await client.SendMessageAsync(sendMessageRequest); Console.WriteLine($Sent a message with id : {response.MessageId});其中messageAttributes通过MessageAttributeValue构造支持String与Number等数据类型Dictionarystring, MessageAttributeValue messageAttributes new Dictionarystring, MessageAttributeValue { { Title, new MessageAttributeValue { DataType String, StringValue The Whistler } }, { Author, new MessageAttributeValue { DataType String, StringValue John Grisham } }, { WeeksOn, new MessageAttributeValue { DataType Number, StringValue 6 } }, };消息属性Message Attributes让消息附带结构化元数据接收方可用于过滤或业务判断。DelaySeconds 10表示消息在队列中延迟 10 秒后才可见。4.5 获取队列 URLGetQueueUrlGetQueueUrl.cs 演示如何按名称查询队列 URL并处理队列不存在的异常var client new AmazonSQSClient(); string queueName New-Example-Queue; try { var response await client.GetQueueUrlAsync(queueName); if (response.HttpStatusCode System.Net.HttpStatusCode.OK) { Console.WriteLine($The URL for {queueName} is: {response.QueueUrl}); } } catch (QueueDoesNotExistException ex) { Console.WriteLine(ex.Message); Console.WriteLine($The queue {queueName} was not found.); }注意代码注释中的提醒如果队列与默认用户的 AWS 区域不同需要在客户端构造函数中显式指定区域。4.6 接收并删除消息ReceiveMessage DeleteMessageReceiveDeleteExample.cs 展示先接收、后删除的完整消费流程var receiveMessageRequest new ReceiveMessageRequest { AttributeNames { SentTimestamp }, MaxNumberOfMessages 1, MessageAttributeNames { All }, QueueUrl queueUrl, VisibilityTimeout 0, WaitTimeSeconds 0, }; var receiveMessageResponse await client.ReceiveMessageAsync(receiveMessageRequest); var deleteMessageRequest new DeleteMessageRequest { QueueUrl queueUrl, ReceiptHandle receiveMessageResponse.Messages[0].ReceiptHandle, }; await client.DeleteMessageAsync(deleteMessageRequest);要点可见性超时VisibilityTimeout接收后消息进入不可见状态防止多个消费者重复处理处理完成必须调用DeleteMessage并携带ReceiptHandle才能删除。MessageAttributeNames { All }表示返回全部消息属性AttributeNames { SentTimestamp }表示返回发送时间戳系统属性。4.7 批量删除消息DeleteMessageBatch与删除队列DeleteQueueSQSWrapper.cs 提供批量删除与队列清理// 批量删除一次请求携带多个 ReceiptHandle var deleteRequest new DeleteMessageBatchRequest() { QueueUrl queueUrl, Entries new ListDeleteMessageBatchRequestEntry() }; foreach (var message in messages) { deleteRequest.Entries.Add(new DeleteMessageBatchRequestEntry() { ReceiptHandle message.ReceiptHandle, Id message.MessageId }); } var deleteResponse await _amazonSQSClient.DeleteMessageBatchAsync(deleteRequest); return deleteResponse.Failed.Any(); // 删除队列 var deleteResponse await _amazonSQSClient.DeleteQueueAsync( new DeleteQueueRequest() { QueueUrl queueUrl }); return deleteResponse.HttpStatusCode HttpStatusCode.OK;DeleteMessageBatch每个条目需提供ReceiptHandle与唯一Id返回结果中Failed列表可用于重试失败条目。批量操作可显著降低 API 调用次数与成本是清理积压消息的首选方式。4.8 接收消息时的长轮询ReceiveMessage场景代码中的 ReceiveMessagesByUrl 展示了长轮询Long Polling配置// Setting WaitTimeSeconds to non-zero enables long polling. var messageResponse await _amazonSQSClient.ReceiveMessageAsync( new ReceiveMessageRequest() { QueueUrl queueUrl, MaxNumberOfMessages maxMessages, WaitTimeSeconds 1 }); return messageResponse.Messages;WaitTimeSeconds设为非零值即启用长轮询当队列为空时请求会挂起等待消息到达从而减少空轮询造成的请求次数与成本。MaxNumberOfMessages单次最多可请求 10 条。五、场景一向队列发布消息Publish messages to queues这是 README 中列出的核心场景代码位于 TopicsAndQueues.cs完整演示 SNS 与 SQS 的集成消息流。该场景可完成以下任务创建主题Topic可选 FIFO 或非 FIFO。创建若干 SQS 队列并订阅到主题可为订阅添加过滤器Filter。向主题发布消息。轮询各队列验证消息是否被正确投递。5.1 场景执行流程从源码注释与RunScenario方法可还原出完整调用链1. CreateTopic —— 创建 SNS 主题FIFO 或非 FIFO 2. CreateQueue —— 创建 SQS 队列 3. GetQueueAttributes —— 获取队列 ARN 4. SetQueueAttributes —— 设置队列策略允许接收 SNS 消息 5. Subscribe —— 将队列订阅到主题 6. Publish —— 向主题发布消息 7. ReceiveMessage —— 轮询队列接收消息 8. DeleteMessageBatch —— 批量删除已处理消息 9. DeleteQueue —— 删除队列 10. Unsubscribe —— 取消订阅 11. DeleteTopic —— 删除主题场景运行时会通过控制台交互引导用户选择是否使用 FIFO 主题、是否启用基于内容的去重Content-based deduplication、是否设置订阅过滤器按tone属性过滤可选值cheerful、funny、serious、sincere等。5.2 FIFO 主题与去重机制场景代码在SetupTopic中详细说明了 FIFO 语义TopicsAndQueues.csFIFO 主题按序投递消息支持去重与消息过滤。使用 FIFO 主题时主题名必须以.fifo结尾。去重 IDDeduplication ID可以显式设置也可以启用内容去重Content-based deduplication由 SNS 对消息内容做哈希自动生成。在 5 分钟去重窗口内相同去重 ID 的消息会被接受但不会重复投递。发布消息时PublishMessagesFIFO 场景强制要求设置消息组 IDMessageGroupId——同一组内的消息按发布顺序接收Console.WriteLine(Because you are using a FIFO topic, you must set a message group ID. \r\nAll messages within the same group will be received in the order they were published.); var messageGroupId GetUserResponse(Enter a message group ID for this message:, 1); if (!_useContentBasedDeduplication) { Console.WriteLine(Because you are not using content-based deduplication, you must enter a deduplication ID.); deduplicationId GetUserResponse(Enter a deduplication ID for this message., 1); } var messageID await SnsWrapper.PublishToTopicWithAttribute( _topicArn, message, tone, toneAttribute, deduplicationId, messageGroupId);5.3 订阅过滤器SetupFilters与CreateFilterPolicy方法演示了如何为队列订阅附加过滤策略TopicsAndQueues.csvar filters new Dictionarystring, Liststring { { tone, filterSelections } }; string filterPolicy JsonSerializer.Serialize(filters);序列化后的过滤策略形如{tone:[cheerful,funny]}只有带匹配tone属性的消息才会被投递到该队列。这实现了同一主题下的消息分流——不同队列可以订阅不同属性的消息。5.4 清理资源场景在结束或异常时都会执行CleanupResources依次询问并删除队列、取消订阅、删除主题避免遗留资源产生持续费用TopicsAndQueues.cs。这也是所有云资源示例应遵循的最佳实践。六、场景二AWS Message Processing Framework for .NET第二个场景演示如何使用 AWS Message Processing Framework for .NET 构建发布与消费 SQS 消息的应用。该框架通过AWS.Messaging命名空间对 SQS 的底层细节轮询、反序列化、路由进行封装开发者只需声明消息类型 → 处理器的映射即可。项目结构包含两个应用与一个测试项目Publisher/Program.cs基于 ASP.NET Core 最小 API 的 Web 应用通过 HTTP 端点接收请求并发布 SQS 消息。Handler/Program.cs命令行应用从 SQS 队列轮询并处理消息。Tests/HandlerTests.cs处理器单元测试。6.1 发布端Publisher发布端通过AddAWSMessageBus注册消息总线并将GreetingMessage类型映射到指定队列和消息标识符builder.Services.AddAWSMessageBus(builder { // Check for input SQS URL. if ((args.Length 1) (args[0].Contains(https://sqs.))) { // 将 GreetingMessage 类型发布到指定队列使用 greetingMessage 标识符 builder.AddSQSPublisherGreetingMessage(args[0], greetingMessage); } });随后定义一个 HTTP 端点调用IMessagePublisher.PublishAsync发布消息Publisher/Program.csapp.MapPost(/greeting, async ([FromServices] IMessagePublisher publisher, Publisher.GreetingMessage message) { return await PostGreeting(message, publisher); }) .WithName(SendGreeting) .WithOpenApi(); public static async TaskIResult PostGreeting(GreetingMessage greetingMessage, IMessagePublisher messagePublisher) { if (greetingMessage.SenderName null || greetingMessage.Greeting null) { return Results.BadRequest(); } await messagePublisher.PublishAsync(greetingMessage); return Results.Ok(); }注意队列 URL 通过命令行参数传入参数需以https://sqs.开头才会注册发布器——这也是框架支持多队列/多类型映射的扩展点。6.2 消费端Handler消费端注册 SQS 轮询器并声明消息类型与处理器的映射关系builder.ConfigureServices(services { services.AddAWSMessageBus(builder { if ((args.Length 1) (args[0].Contains(https://sqs.))) { // 注册轮询指定队列 builder.AddSQSPoller(args[0]); // 将 greetingMessage 类型消息反序列化为 GreetingMessage // 并交给 GreetingMessageHandler 处理 builder.AddMessageHandlerGreetingMessageHandler, GreetingMessage(greetingMessage); } }); });处理器实现IMessageHandlerT接口通过MessageEnvelopeT拿到消息内容后返回处理状态public class GreetingMessageHandler : IMessageHandlerGreetingMessage { public TaskMessageProcessStatus HandleAsync( MessageEnvelopeGreetingMessage messageEnvelope, CancellationToken token default) { Console.WriteLine( $Received message {messageEnvelope.Message.Greeting} from {messageEnvelope.Message.SenderName}); return Task.FromResult(MessageProcessStatus.Success()); } }框架核心价值在于消息路由发布端与消费端使用相同的消息标识符greetingMessage完成解耦无需双方共享任何代码。类型安全消息自动反序列化为强类型对象GreetingMessage无需手动解析 JSON。可测试性IMessageHandler是纯接口抽象单元测试可直接构造MessageEnvelopeT调用处理器。仓库中的 HandlerTests.cs 正是这样做的构造GreetingMessage与MessageEnvelope调用HandleAsync后断言response.IsSuccess为真。七、运行示例与测试7.1 构建与运行通用运行步骤参见 dotnetv3/README.md进入包含.sln文件的目录执行dotnet build SOLUTION.sln。进入包含代码示例与.csproj文件的目录。执行dotnet run运行项目。部分项目包含settings.json文件编译前可按自己的账户与资源修改其中的值也可以添加settings.local.json存放本地配置应用运行时会被自动加载。示例编译后也可以在 IDE 中直接运行。对于需要传入参数的示例如 Message Processing Framework 的 Publisher/Handler通过命令行或 Debug launch profile 传入 SQS 队列 URLdotnet run -- https://sqs.region.amazonaws.com/account-id/queue-name7.2 运行测试⚠️ 运行测试可能产生 AWS 账户费用。测试运行说明同样参见 dotnetv3/README.md。在包含测试项目的目录中执行dotnet test如需更详细的输出dotnet test -l console;verbositydetailed仓库测试按类别区分单元测试与集成测试可单独筛选dotnet test --filter CategoryUnit -l console;verbositydetailed dotnet test --filter CategoryIntegration -l console;verbositydetailed八、代码地图与延伸阅读以下是本文涉及的核心文件索引方便继续深入阅读用途文件SQS 单操作封装创建/属性/策略/接收/批量删除/删队列SQSWrapper.cs入门示例ListQueuesHelloSQS.cs创建队列 发送消息CreateSendExample.cs获取队列 URLGetQueueUrl.cs接收并删除消息ReceiveDeleteExample.csSNS→SQS 消息发布场景TopicsAndQueues.csMessage Processing Framework 发布端Publisher/Program.csMessage Processing Framework 消费端Handler/Program.cs处理器单元测试HandlerTests.cs.NET 示例总览与构建/测试指南dotnetv3/README.md延伸阅读资料Amazon SQS 开发者指南、Amazon SQS API Reference、SDK for .NET 的 SQS API 参考均可在 AWS 官方文档站检索获取。仓库中 SNS 侧的配套封装位于 TopicsAndQueues 跨服务目录可与本文 SQS 操作对照学习完整消息链路。Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. SPDX-License-Identifier: Apache-2.0赞分享示例工程教程后端【免费下载链接】aws-doc-sdk-examplesWelcome to the AWS Code Examples Repository. This repo contains code examples used in the AWS documentation, AWS SDK Developer Guides, and more. For more information, see the Readme.md file below.项目地址https://gitcode.com/gh_mirrors/aw/aws-doc-sdk-examples点击查看免费下载相关推荐AWS SDK for Go 实现 Amazon SQS 队列与消息操作实战指南go/sqs 示例全解AWS SDK for Go 实现 Amazon SQS 队列与消息操作实战指南go/sqs 示例全解 导读 本文围绕 aws doc sdk exampl示例工程教程后端AWS SDK for .NET 操作 Amazon S3 完整指南从基础动作到高级场景实战AWS SDK for .NET 操作 Amazon S3 完整指南从基础动作到高级场景实战 本文以开源仓库 aws doc sdk examples htt示例工程教程后端使用 AWS SDK for C 操作 Amazon SQS完整代码示例与实战指南使用 AWS SDK for C 操作 Amazon SQS完整代码示例与实战指南 导读 本文以 AWS 文档代码示例仓库aws doc sdk exa示例工程教程后端上一篇ReactPage与Web Components集成跨框架编辑器解决方案下一篇Headlamp 前端 KubeCondition 接口全解Kubernetes 状态条件的建模、判断与界面呈现创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考