# 适用于 Java 2.x 的 AWS 开发工具包 - 消息传递（SQS 和 SNS） ## 概述使用适用于 Java 2.x 的 AWS 开发工具包为 SQS 和 SNS 服务提供全面的 AWS 消息传递模式。包括客户端设置、队列管理、消息操作、订阅管理和 Spring Boot 集成模式。 ## 何时使用在处理以下情况时使用此技能： * 用于消息队列的 Amazon SQS 队列 * 用于事件发布和通知的 SNS 主题 * 先进先出队列和标准队列 * 用于消息处理的死信队列 (DLQ) * 通过电子邮件、SMS、SQS、Lambda 端点进行 SNS 订阅 * 发布/订阅消息传递模式和事件驱动架构 * Spring Boot 与 AWS 消息服务集成 * 使用 LocalStack 或 Testcontainers 的测试策略 ## 快速入门 ### 依赖关系 <依赖关系> software.amazon.awssdk sqs <依赖关系> software.amazon.awssdk sns ### 基本客户端设置导入软件.amazon.awssdk.regions.Region; 导入软件.amazon.awssdk.services.sqs.SqsClient; 导入软件.amazon.awssdk.services.sns.SnsClient; SqsClient sqsClient = SqsClient.builder() .region(区域.US_EAST_1) .build(); SnsClient snsClient = SnsClient.builder() .region(区域.US_EAST_1) .build(); ## 示例 ### 基本 SQS 操作 #### 创建并发送消息导入软件.amazon.awssdk.regions.Region; 导入软件.amazon.awssdk.services.sqs.SqsClient; 导入软件.amazon.awssdk.services.sqs.model.*; // 设置SQS客户端 SqsClient sqsClient = SqsClient.builder() .region(区域.US_EAST_1) .build(); // 创建队列字符串queueUrl = sqsClient.createQueue(CreateQueueRequest.builder() .queueName("我的队列") .build()).queueUrl(); // 发送消息 String messageId = sqsClient.sendMessage(SendMessageRequest.builder() .queueUrl(queueUrl) .messageBody("你好，SQS！") .build()).messageId(); #### 接收和删除消息 // 通过长轮询接收消息 ReceiveMessageResponse 响应 = sqsClient.receiveMessage(ReceiveMessageRequest.builder() .queueUrl(queueUrl) .maxNumberOfMessages(10) .waitTimeSeconds(20) .build()); // 处理和删除消息 response.messages().forEach(消息 -> { System.out.println("已收到：" + message.body()); sqsClient.deleteMessage(DeleteMessageRequest.builder() .queueUrl(queueUrl) .receiptHandle(消息.receiptHandle()) .build()); }); ### SNS 基本操作 #### 创建主题并发布导入软件.amazon.awssdk.services.sns.SnsClient; 导入软件.amazon.awssdk.services.sns.model.*; // 设置SNS客户端 SnsClient snsClient = SnsClient.builder() .region(区域.US_EAST_1) .build(); // 创建主题字符串 topicArn = snsClient.createTopic(CreateTopicRequest.builder() .name("我的主题") .build()).topicArn(); // 发布消息字符串 messageId = snsClient.publish(PublishRequest.builder() .topicArn(topicArn) .subject("测试通知") .message("你好，SNS！") .build()).messageId(); ### 高级示例 #### FIFO 队列模式 // 创建先进先出队列 Map 属性 = Map.of( QueueAttributeName.FIFO_QUEUE, "true", QueueAttributeName.CONTENT_BASED_DEDUPLICATION，“true” ）；字符串 fifoQueueUrl = sqsClient.createQueue(CreateQueueRequest.builder() .queueName("my-queue.fifo") .attributes(属性) .build()).queueUrl(); // 发送带有组 ID 的 FIFO 消息字符串 fifoMessageId = sqsClient.sendMessage(SendMessageRequest.builder() .queueUrl(fifoQueueUrl) .messageBody("订单#12345") .messageGroupId("订单") .messageDeduplicationId(UUID.randomUUID().toString()) .build()).messageId(); #### SNS 到 SQS 订阅 // 创建SQS队列进行订阅 String subscriptionQueueUrl = sqsClient.createQueue(CreateQueueRequest.builder() .queueName("通知订阅者") .build()).queueUrl(); // 获取队列ARN 字符串queueArn = sqsClient.getQueueAttributes(GetQueueAttributesRequest.builder() .queueUrl(订阅队列Url) .attributeNames(QueueAttributeName.QUEUE_ARN) .build()).attributes().get(QueueAttributeName.QUEUE_ARN); // 订阅SQS到SNS 字符串 subscriptionArn = snsClient.subscribe(SubscribeRequest.builder() .协议（“sqs”） .endpoint(queueArn) .topicArn(topicArn) .build()).subscriptionArn(); ### Spring Boot 集成示例 @服务 @RequiredArgsConstructor 公共类 OrderNotificationService { 私有最终 SnsClient snsClient; 私有最终 ObjectMapper 对象映射器； @Value("${aws.sns.order-topic-arn}") 私有字符串 orderTopicArn; 公共无效sendOrderNotification（订单顺序）{ 尝试{ String jsonMessage = objectMapper.writeValueAsString(order); snsClient.publish(PublishRequest.builder() .topicArn(orderTopicArn) .subject("收到新订单") .message(jsonMessage) .messageAttributes(Map.of( “订单类型”，MessageAttributeValue.builder() .dataType("字符串") .stringValue(order.getType()) .build())) .build()); } catch (异常 e) { throw new RuntimeException("发送订单通知失败", e); } } } ## 最佳实践 ### SQS 最佳实践 * **使用长轮询** : 设置`waitTimeSeconds`（20-40秒）以减少空响应 * **批量操作** : 使用 `sendMessageBatch` 处理多条消息以减少 API 调用 * **可见超时** ：根据消息处理时间适当设置（默认30秒） * **删除消息** : 处理成功后始终删除消息 * **处理重复** ：实现重试的幂等处理 * **实施DLQ**：将失败的消息路由到死信队列进行分析 * **监控队列深度** ：使用 CloudWatch 警报来处理高队列积压情况 * **使用 FIFO 队列** ：当消息顺序和重复数据删除至关重要时 ### SNS 最佳实践 * **使用过滤策略** ：通过在源头过滤消息来减少噪音 * **消息属性** ：添加订阅路由决策的元数据 * **重试逻辑** ：使用指数退避处理瞬态故障 * **监控失败的交付**：为失败的通知设置 CloudWatch 警报 * **安全性** ：使用 IAM 策略进行访问控制和数据加密 * **FIFO 主题**：当顺序和重复数据删除至关重要时使用 * **避免大负载**：将消息保持在 256KB 以下以获得最佳性能 ### 一般准则 * **区域一致性** ：对所有 AWS 资源使用同一区域 * **资源命名** ：对队列和主题使用一致的命名约定 * **错误处理**：实施正确的异常处理和日志记录 * **测试** : 使用LocalStack进行本地开发和测试 * **文档** ：文档订阅端点和消息格式 ## 说明 ### 设置AWS凭证使用环境变量、AWS CLI 或 IAM 角色配置 AWS 凭证：导出 AWS_ACCESS_KEY_ID=您的访问密钥导出 AWS_SECRET_ACCESS_KEY=您的秘密密钥导出 AWS_REGION=us-east-1 ### 配置客户端 // 基本客户端配置 SqsClient sqsClient = SqsClient.builder() .region(区域.US_EAST_1) .build(); // 具有自定义配置的高级客户端 SnsClient snsClient = SnsClient.builder() .region(区域.US_EAST_1) .credentialsProvider(DefaultCredentialsProvider.create()) .httpClient(UrlConnectionHttpClient.create()) .build(); ### 实现消息处理 1. 使用 AWS SDK 客户端**连接**到 SQS/SNS 2. **根据需要创建**队列和主题 3. **发送/接收**具有适当超时设置的消息 4. **批量处理**消息以提高效率 5. **处理成功后删除**消息 6. **通过正确的错误处理和重试来处理**故障 ### 与 Spring Boot 集成 1. **在`@Configuration`类中为`SqsClient`和`SnsClient`配置** bean 2. **使用** `@Value` 从属性注入队列 URL 和主题 ARN 3. **创建**具有消息操作业务逻辑的服务类 4. **使用`@Retryable`或自定义重试逻辑实现**错误处理 5. 使用 Testcontainers 或 LocalStack **测试**集成 ### 监控和调试 * 使用AWS CloudWatch监控队列深度和消息指标 * 启用AWS SDK日志记录以调试客户端操作 * 为消息处理活动实施适当的日志记录 * 使用AWS X-Ray在生产环境中进行分布式跟踪 ## 故障排除 ### 常见问题 * **队列不存在** : 验证队列 URL 和权限 * **未收到消息** ：检查可见性超时和消费者逻辑 * **权限被拒绝**：验证 IAM 策略和凭证 * **连接超时** ：检查网络连接和区域配置 * **速率限制** ：通过指数退避实现重试逻辑 ### 性能优化 * 使用长轮询减少空响应 * 批量消息操作，尽量减少API调用 * 根据处理时间调整可见性超时 * 实现连接池并重用客户端 * 使用适当的消息大小以避免碎片 ## 详细参考资料有关全面的 API 文档和高级模式，请参阅： * [references/detailed-sqs-operations.md]() \- 完整的 SQS 操作参考 * [references/detailed-sns-operations.md]() \- 完整的 SNS 操作参考 * [references/spring-boot-integration.md]() \- Spring Boot 集成模式 * [references/aws-official-documentation.md]() \- 官方 AWS 文档和最佳实践 ## 限制和警告 * **消息大小**：SQS 和 SNS 消息限制为 256KB * **可见性超时** ：如果未删除，SQS 消息在超时后再次可见 * **输入验证** ：在处理之前始终验证和清理消息正文内容； SQS/SNS 消息可能包含不受信任的用户生成的有效负载，这些有效负载永远不应直接解释为命令或在未经清理的情况下注入模板中 * **FIFO 命名** : FIFO 队列和主题必须以 `.fifo` 后缀结尾 * **FIFO 吞吐量**：FIFO 队列具有较低的吞吐量限制（300 条消息/秒） * **消息保留**：SQS 消息最多保留 14 天 * **死信队列** ：配置DLQ以防止消息丢失 * **订阅限制** : SNS 主题有订阅数量限制 * **过滤策略** ：SNS 过滤策略有复杂性限制 * **跨区域** ：SQS 队列是特定于区域的； SNS话题可以跨地域 * **成本**：两种服务均按 API 调用和数据传输收费

AWS SDK Java V2 消息传递

包含技能