Amazon SNS 消息批处理 - Amazon Simple Notification Service
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅中国的 Amazon Web Services 服务入门

Amazon SNS 消息批处理

什么是消息批处理?

在各个 Publish API 请求中向标准或 FIFO 主题发布消息的替代方法,该方法使用 Amazon SNS PublishBatch API 在单个 API 请求中最多发布 10 条消息。批量发送消息可以帮助您利用 Amazon SNS 将与连接分布式应用程序(A2A 消息收发)或者向人们发送通知(A2P 消息收发)相关的成本最高降低 10 倍。根据您运营所在区域,Amazon SNS 对每秒可以向某个主题发布的消息数量设定了配额。请参阅 Amazon 一般参考指南中的 Amazon SNS 终端节点和配额页面了解有关 API 配额的更多信息。

注意

您在单个 PublishBatch API 请求中发送的所有消息的总合计大小不能超过 262144 字节 (256KB)。

PublishBatch API 将相同的 Publish API 操作用于 IAM 策略。

消息批处理是如何工作的?

使用 PublishBatch API 发布消息与使用 Publish API 发布消息类似。主要的区别在于,需要给 PublishBatch API 请求中的每条消息分配一个唯一的批处理 ID(最多 80 个字符)。这样,Amazon SNS 可以为批处理中的每条消息返回单独的 API 响应,以确认每条消息均已发布或发生故障。对于发布到 FIFO 主题的消息,除了包括分配唯一的批处理 ID 之外,还需要为每条单独的消息包括 MessageDeduplicationIDMessageGroupId

Examples

向标准主题发布一批 10 条的消息

// Imports import com.amazonaws.services.sns.AmazonSNS; import com.amazonaws.services.sns.model.PublishBatchRequest; import com.amazonaws.services.sns.model.PublishBatchRequestEntry; import com.amazonaws.services.sns.model.PublishBatchResult; import com.amazonaws.services.sns.model.AmazonSNSException; import java.util.List; import java.util.stream.Collectors; // Code private static final int MAX_BATCH_SIZE = 10; public static void publishBatchToTopic(AmazonSNS snsClient, String topicArn) { try { // Create the batch entries to send List<PublishBatchRequestEntry> entries = IntStream.range(0, MAX_BATCH_SIZE) .mapToObj(i -> new PublishBatchRequestEntry() .withId("id" + i) .withMessage("message" + i)) .collect(Collectors.toList()); // Create the batch request PublishBatchRequest request = new PublishBatchRequest() .withTopicArn(topicArn) .withPublishBatchRequestEntries(entries); // Publish the batch request PublishBatchResult publishBatchResult = snsClient.publishBatch(request); // Handle the successfully sent messages publishBatchResult.getSuccessful().forEach(publishBatchResultEntry -> { System.out.println("Batch Id for successful message: " + publishBatchResultEntry.getId()); System.out.println("Message Id for successful message: " + publishBatchResultEntry.getMessageId()); }); // Handle the failed messages publishBatchResult.getFailed().forEach(batchResultErrorEntry -> { System.out.println("Batch Id for failed message: " + batchResultErrorEntry.getId()); System.out.println("Error Code for failed message: " + batchResultErrorEntry.getCode()); System.out.println("Sender Fault for failed message: " + batchResultErrorEntry.getSenderFault()); System.out.println("Failure Message for failed message: " + batchResultErrorEntry.getMessage()); }); } catch (AmazonSNSException e) { // Handle any exceptions from the request System.err.println(e.getMessage()); System.exit(1); } }

向 FIFO 主题发布一批 10 条的消息

// Imports import com.amazonaws.services.sns.AmazonSNS; import com.amazonaws.services.sns.model.PublishBatchRequest; import com.amazonaws.services.sns.model.PublishBatchRequestEntry; import com.amazonaws.services.sns.model.PublishBatchResult; import com.amazonaws.services.sns.model.AmazonSNSException; import java.util.List; import java.util.stream.Collectors; // Code private static final int MAX_BATCH_SIZE = 10; public static void publishBatchToFifoTopic(AmazonSNS snsClient, String topicArn) { try { // Create the batch entries to send List<PublishBatchRequestEntry> entries = IntStream.range(0, MAX_BATCH_SIZE) .mapToObj(i -> new PublishBatchRequestEntry() .withId("id" + i) .withMessage("message" + i) .withMessageGroupId("groupId") .withMessageDeduplicationId("deduplicationId" + i)) .collect(Collectors.toList()); // Create the batch request PublishBatchRequest request = new PublishBatchRequest() .withTopicArn(topicArn) .withPublishBatchRequestEntries(entries); // Publish the batch request PublishBatchResult publishBatchResult = snsClient.publishBatch(request); // Handle the successfully sent messages publishBatchResult.getSuccessful().forEach(publishBatchResultEntry -> { System.out.println("Batch Id for successful message: " + publishBatchResultEntry.getId()); System.out.println("Message Id for successful message: " + publishBatchResultEntry.getMessageId()); System.out.println("SequenceNumber for successful message: " + publishBatchResultEntry.getSequenceNumber()); }); // Handle the failed messages publishBatchResult.getFailed().forEach(batchResultErrorEntry -> { System.out.println("Batch Id for failed message: " + batchResultErrorEntry.getId()); System.out.println("Error Code for failed message: " + batchResultErrorEntry.getCode()); System.out.println("Sender Fault for failed message: " + batchResultErrorEntry.getSenderFault()); System.out.println("Failure Message for failed message: " + batchResultErrorEntry.getMessage()); }); } catch (AmazonSNSException e) { // Handle any exceptions from the request System.err.println(e.getMessage()); System.exit(1); } }