使用 AWS SDK for Kotlin 构建 SNS 主题与 SQS 队列的发布/订阅、过滤与 FIFO 实战指南

示例工程教程后端【免费下载链接】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 官方代码示例仓库中kotlin/usecases/topics_and_queues示例的完整技术指南。该示例基于 AWS SDK for Kotlin通过命令行交互程序演示如何创建 Amazon SNS 主题、将 Amazon SQS 队列订阅到主题、启用 FIFO先进先出队列、配置基于属性的过滤订阅并发布/接收消息的完整链路。阅读本文后你将掌握 SNS 与 SQS 组合使用时的核心配置逻辑、FIFO 去重与消息分组的原理以及使用 Kotlin 调用SnsClient与SqsClient的完整代码实现。一、发布/订阅架构为什么需要 SNS SQS 组合发布/订阅Publish and Subscribe是一种信息传递机制广泛用于社交媒体以及软件内部的模块解耦。生产者Producer发布消息订阅者Subscriber接收消息。在这一模型中消息生产者与消息消费者被解耦因此消息传递更加灵活、健壮。该示例不会构建一个完整的端到端应用程序而是提供一个可以亲手把玩发布/订阅架构的交互式命令行程序帮助你理解两种 AWS 消息服务的分工差异Amazon SNS 是推送Push服务它主动将消息推送到各类端点如电子邮件地址、移动应用端点或 SQS 队列。生产者只负责把消息发布到主题Topic由 SNS 负责投递。Amazon SQS 是轮询Poll服务订阅者通过调用接收消息APIReceiveMessage主动轮询队列。任何代码都可以轮询队列消息会一直保存在队列中直到你显式删除它这给消息的处理方式带来了更大的灵活性。单独使用 SNS 即可实现发布/订阅但将 SNS 与 SQS 组合使用可以获得更灵活的消息消费方式SNS 负责推送与过滤SQS 负责缓存与异步消费。二、FIFO 主题与队列的核心概念FIFO先进先出主题示例程序启动后首先询问是否使用 FIFO 主题Would you like to work with FIFO topics? (y/n)FIFO 主题在创建时配置启用后还会解锁其他选项去重、消息分组、消息过滤等。FIFO 主题保证消息严格有序投递。消息去重DeduplicationUse content-based deduplication instead of a deduplication ID? (y/n)去重仅对 FIFO 主题可用。去重机制防止订阅者针对被判定为重复的事件响应多次。规则如下若一条消息发布到 SNS FIFO 主题后在5 分钟去重时间窗口内发现与已有消息具有相同的去重 ID则该消息会被接收但不会被投递。基于内容的去重Content-based deduplication使用消息内容的哈希值作为去重 ID。选择此项后发布消息时无需手动提供去重 ID。显式去重 ID若未启用基于内容的去重则每条消息都必须携带去重 IDDeduplication ID。消息分组Message Grouping对于 FIFO 主题每条消息必须携带消息组 IDGroup ID。组 ID 决定消息在主题内的排序范围相同组 ID 的消息按发布顺序严格排列。组 ID 最多 128 个字符可包含字母数字字符(a-z, A-Z, 0-9)以及标点符号(!#$%()*,-./:;?[\]^_{|}~)。消息去重 ID 的字符限制与组 ID 相同同样最多 128 个字符。命名规则示例对主题与队列名称有明确约束对象长度允许字符特殊规则SNS 主题名称1–256 字符大小写 ASCII 字母、数字、下划线、连字符选择 FIFO 后程序自动追加.fifo后缀FIFO 主题必需SQS 队列名称1–80 字符大小写 ASCII 字母、数字、下划线、连字符选择 FIFO 后程序自动追加.fifo后缀FIFO 队列必需三、示例的完整交互流程程序通过标准输入Scanner逐项收集配置全程共 11 个步骤。以下按 README 与实际源码 SNSWorkflow.kt 还原完整界面。1. 创建 SNS 主题Enter a name for your SNS topic:若前面选择了 FIFO 主题程序会自动在名称后追加.fifo。选择 FIFO 后程序会立即询问去重方式并收集组 ID / 去重 ID源码第 90–113 行选择基于内容去重y提示Enter a group id value选择显式去重 IDn依次提示Enter deduplication Id value与Enter a group id value2. 创建 SQS 队列Enter a name for an SQS queue.为每个订阅者单独建队列是有益的——你可以为不同订阅者定制消息消费方式与消息过滤规则。FIFO 场景下队列名称同样自动追加.fifo后缀。3. 为订阅添加过滤仅 FIFO 场景Filter messages for queue name.fifos subscription to the topic topic name.fifo? (y/n)若选择为订阅添加过滤器则按预定属性集合过滤You can filter messages by one or more of the following tone attributes. 1. cheerful 2. funny 3. serious 4. sincere Enter a number (or enter zero to stop adding more).可连续选择多个tone属性输入0结束选择。4. 发布消息Enter a message text to publish.之后进入发布阶段FIFO 场景询问是否为消息附加tone属性Add an attribute to this message? (y/n)随后从cheerful / funny / serious / sincere中选择一个发布到主题。README 中描述了重复发布选项Post another message? (y/n)允许连续发布多条消息当前仓库源码的main流程在发布一条消息后即进入接收阶段SNSWorkflow.kt 第 207–234 行读者可按需自行扩展循环发布逻辑。5. 接收、显示并清理发布结束后程序轮询队列并显示消息内容Message Id 与完整 Body随后依次删除已接收消息、退订、删除队列与主题完成资源的完整回收。四、源码实现深度解析4.1 创建主题FIFO 与非 FIFO 的分支非 FIFO 主题直接通过CreateTopicRequest创建SNSWorkflow.kt 第 568–577 行suspend fun createSNSTopic(topicName: String?): String? { val request CreateTopicRequest { name topicName } SnsClient { region us-east-1 }.use { snsClient - val result snsClient.createTopic(request) return result.topicArn } }FIFO 主题则在CreateTopicRequest中通过attributes声明两个关键属性第 579–597 行suspend fun createFIFO(topicName: String?, duplication: String): String? { val topicAttributes: MutableMapString, String HashMap() if (duplication.compareTo(n) 0) { topicAttributes[FifoTopic] true topicAttributes[ContentBasedDeduplication] false } else { topicAttributes[FifoTopic] true topicAttributes[ContentBasedDeduplication] true } ... }主题属性值含义FifoTopictrue启用 FIFO消息严格有序ContentBasedDeduplicationtrue/false是否基于消息内容哈希自动去重可见duplication选项直接映射到ContentBasedDeduplication属性这就是选择内容去重后无需再输入去重 ID的底层原因。4.2 创建队列FIFO 队列属性createQueue根据selectFIFO分支创建队列第 527–566 行。FIFO 队列通过QueueAttributeName.FifoQueue属性声明if (selectFIFO) { val attrs mutableMapOfString, String() attrs[QueueAttributeName.FifoQueue.toString()] true val createQueueRequest CreateQueueRequest { queueName queueNameVal attributes attrs } ... }创建成功后通过GetQueueUrlRequest获取队列 URL后续所有 SQS 操作都以queueUrl为寻址依据。4.3 获取队列 ARN 并附加 IAM 策略getSQSQueueAttrs通过GetQueueAttributesRequest请求QueueAttributeName.QueueArn属性得到队列 ARN第 506–525 行。要让 SNS 能把消息投递到 SQS 队列必须为队列附加一条 IAM 资源策略SetQueueAttributesRequestPolicy属性第 491–504 行。源码中策略内容如下{ Statement: [ { Effect: Allow, Principal: { Service: sns.amazonaws.com }, Action: sqs:SendMessage, Resource: SQS 队列 ARN, Condition: { ArnEquals: { aws:SourceArn: SNS 主题 ARN } } } ] }策略要点Principal限定为sns.amazonaws.com即只允许 SNS 服务访问Action仅授予sqs:SendMessage最小权限Condition通过ArnEquals将来源限定为当前主题的 ARN防止其他主题向该队列投递。测试代码 AWSSNSTest.kt 在非 FIFO 用例中展示了等价的 ARN 拼接写法arn:aws:sqs:us-east-1:accountId:queueName。4.4 订阅队列并配置过滤策略subQueue是示例的核心方法第 437–489 行通过SubscribeRequest完成订阅request SubscribeRequest { protocol sqs endpoint queueArnVal returnSubscriptionArn true topicArn topicArnVal }关键参数protocol sqs声明订阅端点是 SQS 队列endpoint queueArnVal目标队列的 ARNreturnSubscriptionArn true立即返回订阅 ARN供后续退订使用。当用户选择了tone过滤属性时示例通过 Gson 构造{tone: [...]}JSON再调用setSubscriptionAttributes将FilterPolicy写入订阅第 468–485 行val attributeNameVal FilterPolicy val jsonString {\tone\: []} val jsonObject gson.fromJson(jsonString, JsonObject::class.java) val toneArray jsonObject.getAsJsonArray(tone) for (value: String? in filterList) { toneArray.add(JsonPrimitive(value)) } val updatedJsonString: String gson.toJson(jsonObject) val attRequest SetSubscriptionAttributesRequest { subscriptionArn result.subscriptionArn attributeName attributeNameVal attributeValue updatedJsonString } snsClient.setSubscriptionAttributes(attRequest)例如选择cheerful与funny后最终写入的过滤策略为{tone:[cheerful,funny]}。此后只有携带匹配tone消息属性的消息才会被投递到该队列——过滤由 SNS 在服务端完成队列只会收到符合规则的消息。4.5 发布消息普通与 FIFO 的差异普通主题的发布第 353–363 行val request PublishRequest { message messageVal topicArn topicArnVal }FIFO 主题的发布pubMessageFIFO第 365–434 行则依据是否使用内容去重 / 是否附加消息属性组合出四种分支内容去重 无属性携带messageGroupId即可显式去重 ID 无属性携带messageDeduplicationId与messageGroupId内容去重 附加属性携带messageGroupId注意当前源码此分支未把messageAttributes附加到请求中若你的业务依赖消息属性过滤可留意这一细节显式去重 ID 附加属性同时携带messageDeduplicationId、messageGroupId与messageAttributes。消息属性通过MessageAttributeValue构造声明数据类型为Stringval messAttr aws.sdk.kotlin.services.sns.model.MessageAttributeValue { dataType String stringValue true } val mapAtt: MapString, aws.sdk.kotlin.services.sns.model.MessageAttributeValue mapOf(msgAttValue to messAttr)mapOf(msgAttValue to messAttr)中的键如cheerful就是过滤策略匹配的依据与订阅端FilterPolicy中的tone数组值一一对应。4.6 接收与删除消息receiveMessages第 332–351 行通过ReceiveMessageRequest轮询队列maxNumberOfMessages 5当消息携带属性msgAttValue非空时额外设置waitTimeSeconds 1启用短轮询等待val receiveRequest ReceiveMessageRequest { queueUrl queueUrlVal waitTimeSeconds 1 maxNumberOfMessages 5 }收到消息后deleteMessages第 312–330 行将MessageId组装为DeleteMessageBatchRequestEntry通过deleteMessageBatch批量删除避免消息被重复消费。4.7 收尾清理示例在流程末尾依次执行第 275–310 行unSub调用UnsubscribeRequest退订deleteSQSQueue先按队列名解析 URL再deleteQueuedeleteSNSTopic调用DeleteTopicRequest删除主题。整个生命周期从创建到删除闭环可安全反复运行。五、构建与运行构建配置示例使用 Gradle 管理依赖构建脚本位于 kotlin/usecases/topics_and_queues/build.gradle.kts核心配置如下plugins { kotlin(jvm) version 1.9.0 application } dependencies { implementation(aws.sdk.kotlin:sns:1.0.0) implementation(aws.sdk.kotlin:sqs:1.0.0) implementation(aws.smithy.kotlin:http-client-engine-okhttp:0.30.0) implementation(aws.smithy.kotlin:http-client-engine-crt:0.30.0) implementation(com.google.code.gson:gson:2.10.1) testImplementation(org.junit.jupiter:junit-jupiter:5.9.2) implementation(org.jetbrains.kotlinx:kotlinx-coroutines-core:1.7.1) }依赖说明aws.sdk.kotlin:sns/aws.sdk.kotlin:sqsAWS SDK for Kotlin 的 SNS、SQS 客户端http-client-engine-okhttp/http-client-engine-crtSmithy Kotlin 提供的 HTTP 传输引擎CRT 引擎支持更长超时与流式场景gson用于构造FilterPolicyJSONkotlinx-coroutines-coreSDK 的 suspend API 依赖协程测试框架为 JUnit 5useJUnitPlatform()JVM 目标版本为 17。前置条件与运行运行前需要完成环境准备配置 AWS 凭据SnsClient与SqsClient均显式指定region us-east-1并确保 IAM 权限覆盖 Amazon SNS 与 Amazon SQS 的对应操作。在kotlin/usecases/topics_and_queues目录下执行gradle run程序启动后按上文交互流程逐步操作即可。注意创建主题、队列等操作会在你的 AWS 账号中产生实际资源与费用示例运行结束后会自动完成清理。六、自动化测试验证示例附带完整的端到端测试 AWSSNSTest.kt通过 JUnit 5 按Order顺序执行两个用例testWorkflowFIFO走 FIFO 全链路——创建 FIFO 主题显式去重 IDdup100、组 IDgroup100、创建 FIFO 队列、附加 IAM 策略、订阅、发布消息、接收、删除、退订、清理testWorkflowNonFIFO走标准主题链路验证非 FIFO 场景下的同名操作。测试用(1..10000).random()生成随机后缀如topic123.fifo、queue456.fifo避免资源命名冲突发布后使用delay(1000)等待消息投递再轮询队列断言接收结果。这两个用例同时验证了 README 中两种分支路径的代码可运行性。执行测试gradle test七、注意事项运行该代码可能产生 AWS 账号费用创建主题、队列及消息投递均按量计费运行测试同样可能产生费用建议为代码授予最小权限Least Privilege仅授予完成任务所需的最低权限。本示例的策略即遵循此原则只授予sqs:SendMessage且限定来源 ARN该代码未经所有 AWS 区域验证如需在生产区域运行请以 AWS 区域服务可用性为准。延伸阅读本示例对应的入门文档kotlin/usecases/topics_and_queues/README.md完整主流程实现SNSWorkflow.kt端到端测试用例AWSSNSTest.ktGradle 构建脚本build.gradle.kts同一仓库中面向 .NET 的等价示例dotnetv3/cross-service/TopicsAndQueues 可作跨语言对照参考赞分享示例工程教程后端【免费下载链接】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 C 实战使用 SNS 主题与 SQS 队列实现带过滤与 FIFO 的发布订阅AWS SDK for C 实战使用 SNS 主题与 SQS 队列实现带过滤与 FIFO 的发布订阅 本篇文章基于 aws doc sdk example示例工程教程后端AWS .NET 示例用 SNS 主题 SQS 队列实现发布订阅、FIFO 去重与过滤实战TopicsAndQueuesAWS .NET 示例用 SNS 主题 SQS 队列实现发布订阅、FIFO 去重与过滤实战TopicsAndQueues 导读 本文基于 AWS 官方示例工程教程后端使用 Amazon SNS 主题与 Amazon SQS 队列实现发布订阅Go 版 FIFO 与消息过滤实战指南使用 Amazon SNS 主题与 Amazon SQS 队列实现发布订阅Go 版 FIFO 与消息过滤实战指南 导读 本文基于 AWS SDK for Go示例工程教程后端上一篇Pokerogue安全最佳实践防篡改与数据验证策略下一篇如何快速构建企业级网页数据平台技术栈选择的终极指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考