aws-sdk-java-v2-messaging

AWS SDK for Java 2.x - 消息传递（SQS 和 SNS）

概述

介绍如何使用 AWS SDK for Java 2.x 处理 SQS 队列和 SNS 主题：客户端设置、队列管理、消息操作、订阅以及 Spring Boot 集成。

适用场景

• 设置 SQS 队列（标准队列或 FIFO 队列）以缓冲消息
• 使用 SNS 主题和订阅实现发布/订阅
• 使用长轮询处理 SQS 队列中的消息
• 配置死信队列（DLQ）以处理错误
• 将 AWS 消息传递集成到 Spring Boot 应用程序中
• 使用 SQS/SNS 构建事件驱动架构

示例

快速设置

依赖项：

<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>sqs</artifactId>
</dependency>
<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>sns</artifactId>
</dependency>

客户端配置：

SqsClient sqsClient = SqsClient.builder()
    .region(Region.US_EAST_1)
    .credentialsProvider(DefaultCredentialsProvider.create())
    .build();

SnsClient snsClient = SnsClient.builder()
    .region(Region.US_EAST_1)
    .build();

SQS 操作

创建并发送消息：

String queueUrl = sqsClient.createQueue(CreateQueueRequest.builder()
    .queueName("my-queue")
    .build()).queueUrl();

String messageId = sqsClient.sendMessage(SendMessageRequest.builder()
    .queueUrl(queueUrl)
    .messageBody("Hello, SQS!")
    .build()).messageId();

接收并删除消息：

ReceiveMessageResponse response = sqsClient.receiveMessage(ReceiveMessageRequest.builder()
    .queueUrl(queueUrl)
    .maxNumberOfMessages(10)
    .waitTimeSeconds(20)
    .build());

response.messages().forEach(message -> {
    processMessage(message.body());
    sqsClient.deleteMessage(DeleteMessageRequest.builder()
        .queueUrl(queueUrl)
        .receiptHandle(message.receiptHandle())
        .build());
});

FIFO 队列：

Map<QueueAttributeName, String> attributes = Map.of(
    QueueAttributeName.FIFO_QUEUE, "true",
    QueueAttributeName.CONTENT_BASED_DEDUPLICATION, "true"
);

String fifoQueueUrl = sqsClient.createQueue(CreateQueueRequest.builder()
    .queueName("my-queue.fifo")
    .attributes(attributes)
    .build()).queueUrl();

sqsClient.sendMessage(SendMessageRequest.builder()
    .queueUrl(fifoQueueUrl)
    .messageBody("Order #12345")
    .messageGroupId("orders")
    .messageDeduplicationId(UUID.randomUUID().toString())
    .build());

SNS 操作

创建主题并发布：

String topicArn = snsClient.createTopic(CreateTopicRequest.builder()
    .name("my-topic")
    .build()).topicArn();

snsClient.publish(PublishRequest.builder()
    .topicArn(topicArn)
    .subject("Test Notification")
    .message("Hello, SNS!")
    .build());

SNS 到 SQS 订阅：

String queueArn = sqsClient.getQueueAttributes(GetQueueAttributesRequest.builder()
    .queueUrl(queueUrl)
    .attributeNames(QueueAttributeName.QUEUE_ARN)
    .build()).attributes().get(QueueAttributeName.QUEUE_ARN);

snsClient.subscribe(SubscribeRequest.builder()
    .protocol("sqs")
    .endpoint(queueArn)
    .topicArn(topicArn)
    .build());

Spring Boot 集成

@Service
@RequiredArgsConstructor
public class OrderNotificationService {
    private final SnsClient snsClient;
    private final ObjectMapper objectMapper;

    @Value("${aws.sns.order-topic-arn}")
    private String orderTopicArn;

    public void sendOrderNotification(Order order) throws JsonProcessingException {
        snsClient.publish(PublishRequest.builder()
            .topicArn(orderTopicArn)
            .subject("New Order Received")
            .message(objectMapper.writeValueAsString(order))
            .messageAttributes(Map.of(
                "orderType", MessageAttributeValue.builder()
                    .dataType("String")
                    .stringValue(order.getType())
                    .build()))
            .build());
    }
}

操作说明

实现消息处理（含验证）

1. 创建队列/主题，并进行适当配置
2. 发送消息，并验证是否返回 messageId
3. 接收消息，并使用长轮询（waitTimeSeconds: 20）
4. 处理消息——处理前验证有效负载
5. 仅在成功处理后删除消息——验证删除响应
6. 定期检查 DLQ，使用 redrivePolicy 检查失败消息
7. 验证投递情况——监控 CloudWatch 的 NumberOfMessagesSent 指标

验证清单：

// After send
if (messageId == null || messageId.isEmpty()) {
    throw new MessagingException("Message send failed - no messageId returned");
}

// After receive
if (response.messages().isEmpty()) {
    log.debug("No messages available - normal with long polling");
}

// After delete
if (!deleteResponse.sdkHttpResponse().isSuccessful()) {
    throw new MessagingException("Message deletion failed");
}

设置凭证

export AWS_ACCESS_KEY_ID=your-access-key
export AWS_SECRET_ACCESS_KEY=your-secret-key
export AWS_REGION=us-east-1

监控和调试

• CloudWatch 指标：ApproximateNumberOfMessages、NumberOfMessagesSent、NumberOfMessagesReceived
• 启用 SDK 日志记录：将 software.amazon.awssdk 设为 DEBUG 级别
• 使用 X-Ray 进行分布式跟踪

最佳实践

SQS：

• 使用长轮询（20-40 秒）以减少空响应和成本
• 成功处理消息后务必删除
• 实现幂等处理以应对重复消息
• 为失败消息配置 DLQ（redrivePolicy）
• 在顺序很重要时使用 FIFO 队列（上限为 300 条消息/秒）

SNS：

• 使用筛选策略减少不必要的投递
• 消息大小保持在 256KB 以内
• 使用指数退避实现重试
• 监控 NumberOfNotificationFailed 指标

通用：

• 使用 IAM 角色，而非静态凭证
• 复用客户端（客户端是线程安全的）
• 使用 LocalStack 或 Testcontainers 进行测试

详细参考

• references/detailed-sqs-operations.md
• references/detailed-sns-operations.md
• references/spring-boot-integration.md
• references/aws-official-documentation.md

限制和警告

• 消息大小：SQS 和 SNS 的上限为 256KB
• 可见性超时：未删除的消息会在超时后重新出现——处理后务必删除
• 输入验证：处理前清理消息正文——消息可能包含不可信的有效负载
• FIFO 命名：名称必须以 .fifo 后缀结尾
• FIFO 吞吐量：每个队列为 300 条消息/秒（如需更高吞吐量，请使用分区）
• 消息保留：SQS 最多保留消息 14 天
• 必须配置 DLQ：配置死信队列以防止消息丢失
• 区域限制：SQS 队列仅限所在区域；跨区域需使用 SNS