本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # 发布 Amazon SNS 消息在[创建 Amazon SNS 主题](sns-create-topic.md)并为终端节点[订阅](sns-create-subscribe-endpoint-to-topic.md)主题后，可以将消息*发布*到主题。发布消息时，Amazon SNS 会尝试将消息传输给订阅的[终端节点](sns-create-subscribe-endpoint-to-topic.md#sns-endpoints)。 ## 要使用 Amazon SNS 主题发布消息 AWS 管理控制台 1. 登录 [Amazon SNS 控制台](https://console.aws.amazon.com/sns/home)。 1. 在左侧导航窗格中，选择**主题**。 1. 在 **Topics**（主题）页上，选择一个主题，然后选择**Publish message**（发布主题）。控制台将打开 **Publish message to topic**（将消息发布到主题）页面。 1. 在 **Message details（消息详细信息）**部分中，执行以下操作： 1. （可选）输入消息 **Subject**（主题）。 1. 对于 [FIFO topic](sns-fifo-topics.md)（FIFO 主题），输入 **Message group ID**（消息组 ID）。FIFO 主题需要**消息组 ID**。同一消息组中的消息按消息的发布顺序传输。 1. （可选）对于标准主题，请输入**消息组 ID**。此 ID 会转发给所有 Amazon SQS 标准订阅，不用于或发送到任何其他端点类型。 1. 对于 FIFO 主题，请输入 **Message deduplication ID**（消息重复数据删除 ID）。如果您为主题启用了 **Content-based message deduplication**（基于内容的消息重复数据删除）设置，则此 ID 为可选项。 1. （可选）对于 [mobile push notifications](sns-ttl.md)（移动推送通知），输入 **Time to Live (TTL)**（存活时间 (TTL)）值（以秒为单位）。这是推送通知服务（例如 Apple 推送通知服务 (APNs) 或 Firebase Cloud Messaging (FCM)）将消息传递到端点所需的时间。 1. 在 **Message body (消息正文)** 部分中，执行以下操作之一： 1. 选择 **Identical payload for all delivery protocols（完全相同负载用于所有交付协议）**，然后输入消息。 1. 选择 **Custom payload for each delivery protocol（对每个交付协议使用自定义负载）**，然后输入 JSON 对象定义要发送给每个协议的消息。有关更多信息，请参阅 [使用特定于平台的有效载荷发布 Amazon SNS 通知](sns-send-custom-platform-specific-payloads-mobile-devices.md)。 1. 在 **Message attributes（消息属性）**部分中，添加您希望 Amazon SNS 与订阅属性 `FilterPolicy` 相匹配的任何属性，以确定订阅的终端节点是否对发布的消息感兴趣。 1. 对于 **Type**（类型），选择属性类型，例如 **String.Array**。 **注意** 对于属性类型 **String.Array**，请将该数组放入方括号 (`[]`) 内。在该数组内，将字符串值加入双引号内。数字以及关键字 `true`、`false` 和 `null` 无需加引号。 1. 输入属性**名称**，例如 `customer_interests`。 1. 输入属性**值**，例如 `["soccer", "rugby", "hockey"]`。如果属性类型为 **String**、**String.Array** 或 **Number**，Amazon SNS 会首先依据订阅的[筛选策略](sns-message-filtering.md)（如果存在）来评估该消息属性，然后再将消息发送至该订阅，前提是筛选策略范围没有明确设置为 `MessageBody`。有关更多信息，请参阅 [Amazon SNS 消息属性](sns-message-attributes.md)。 1. 选择**发布消息**。消息将发布到主题，且控制台将打开主题的 **Details**（详细信息）页面。 ## 使用 AWS SDK 向主题发布消息要使用 S AWS DK，必须使用您的凭据对其进行配置。有关更多信息，请参阅[和*工具参考指南中的共享配置AWS SDKs 和*凭据文件](https://docs.aws.amazon.com/sdkref/latest/guide/creds-config-files.html)。以下代码示例演示如何使用 `Publish`。 ------ #### [ .NET ] **适用于 .NET 的 SDK** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/dotnetv3/SNS#code-examples)中查找完整示例，了解如何进行设置和运行。向主题发布消息。 ``` using System; using System.Threading.Tasks; using Amazon.SimpleNotificationService; using Amazon.SimpleNotificationService.Model; ///

/// This example publishes a message to an Amazon Simple Notification /// Service (Amazon SNS) topic. ///

public class PublishToSNSTopic { public static async Task Main() { string topicArn = "arn:aws:sns:us-east-2:000000000000:ExampleSNSTopic"; string messageText = "This is an example message to publish to the ExampleSNSTopic."; IAmazonSimpleNotificationService client = new AmazonSimpleNotificationServiceClient(); await PublishToTopicAsync(client, topicArn, messageText); } ///

/// Publishes a message to an Amazon SNS topic. ///

/// The initialized client object used to publish /// to the Amazon SNS topic. /// The ARN of the topic. /// The text of the message. public static async Task PublishToTopicAsync( IAmazonSimpleNotificationService client, string topicArn, string messageText) { var request = new PublishRequest { TopicArn = topicArn, Message = messageText, }; var response = await client.PublishAsync(request); Console.WriteLine($"Successfully published message ID: {response.MessageId}"); } } ``` 使用组、复制和属性选项向主题发布消息。 ``` ///

/// Publish messages using user settings. ///

/// Async task. public static async Task PublishMessages() { Console.WriteLine("Now we can publish messages."); var keepSendingMessages = true; string? deduplicationId = null; string? toneAttribute = null; while (keepSendingMessages) { Console.WriteLine(); var message = GetUserResponse("Enter a message to publish.", "This is a sample message"); if (_useFifoTopic) { 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."); Console.WriteLine(); 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."); Console.WriteLine("Enter a deduplication ID for this message."); deduplicationId = GetUserResponse("Enter a deduplication ID for this message.", "1"); } if (GetYesNoResponse("Add an attribute to this message?")) { Console.WriteLine("Enter a number for an attribute."); for (int i = 0; i < _tones.Length; i++) { Console.WriteLine($"\t{i + 1}. {_tones[i]}"); } var selection = GetUserResponse("", "1"); int.TryParse(selection, out var selectionNumber); if (selectionNumber > 0 && selectionNumber < _tones.Length) { toneAttribute = _tones[selectionNumber - 1]; } } var messageID = await SnsWrapper.PublishToTopicWithAttribute( _topicArn, message, "tone", toneAttribute, deduplicationId, messageGroupId); Console.WriteLine($"Message published with id {messageID}."); } keepSendingMessages = GetYesNoResponse("Send another message?", false); } } ``` 将用户的选择应用于发布操作。 ``` ///

/// Publish a message to a topic with an attribute and optional deduplication and group IDs. ///

/// The ARN of the topic. /// The message to publish. /// The optional attribute for the message. /// The optional attribute value for the message. /// The optional deduplication ID for the message. /// The optional group ID for the message. /// The ID of the message published. public async Task PublishToTopicWithAttribute( string topicArn, string message, string? attributeName = null, string? attributeValue = null, string? deduplicationId = null, string? groupId = null) { var publishRequest = new PublishRequest() { TopicArn = topicArn, Message = message, MessageDeduplicationId = deduplicationId, MessageGroupId = groupId }; if (attributeValue != null) { // Add the string attribute if it exists. publishRequest.MessageAttributes = new Dictionary { { attributeName!, new MessageAttributeValue() { StringValue = attributeValue, DataType = "String"} } }; } var publishResponse = await _amazonSNSClient.PublishAsync(publishRequest); return publishResponse.MessageId; } ``` + 有关 API 详细信息，请参阅《适用于 .NET 的 AWS SDK API Reference》**中的 [Publish](https://docs.aws.amazon.com/goto/DotNetSDKV3/sns-2010-03-31/Publish)。 ------ #### [ C\$1\$1 ] **SDK for C\$1\$1** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/cpp/example_code/sns#code-examples)中查找完整示例，了解如何进行设置和运行。 ``` //! Send a message to an Amazon Simple Notification Service (Amazon SNS) topic. /*! \param message: The message to publish. \param topicARN: The Amazon Resource Name (ARN) for an Amazon SNS topic. \param clientConfiguration: AWS client configuration. \return bool: Function succeeded. */ bool AwsDoc::SNS::publishToTopic(const Aws::String &message, const Aws::String &topicARN, const Aws::Client::ClientConfiguration &clientConfiguration) { Aws::SNS::SNSClient snsClient(clientConfiguration); Aws::SNS::Model::PublishRequest request; request.SetMessage(message); request.SetTopicArn(topicARN); const Aws::SNS::Model::PublishOutcome outcome = snsClient.Publish(request); if (outcome.IsSuccess()) { std::cout << "Message published successfully with id '" << outcome.GetResult().GetMessageId() << "'." << std::endl; } else { std::cerr << "Error while publishing message " << outcome.GetError().GetMessage() << std::endl; } return outcome.IsSuccess(); } ``` 发布带有属性的消息。 ``` static const Aws::String TONE_ATTRIBUTE("tone"); static const Aws::Vector TONES = {"cheerful", "funny", "serious", "sincere"}; Aws::Client::ClientConfiguration clientConfig; // Optional: Set to the AWS Region (overrides config file). // clientConfig.region = "us-east-1"; Aws::SNS::SNSClient snsClient(clientConfiguration); Aws::SNS::Model::PublishRequest request; request.SetTopicArn(topicARN); Aws::String message = askQuestion("Enter a message text to publish. "); request.SetMessage(message); if (filteringMessages && askYesNoQuestion( "Add an attribute to this message? (y/n) ")) { for (size_t i = 0; i < TONES.size(); ++i) { std::cout << " " << (i + 1) << ". " << TONES[i] << std::endl; } int selection = askQuestionForIntRange( "Enter a number for an attribute. ", 1, static_cast(TONES.size())); Aws::SNS::Model::MessageAttributeValue messageAttributeValue; messageAttributeValue.SetDataType("String"); messageAttributeValue.SetStringValue(TONES[selection - 1]); request.AddMessageAttributes(TONE_ATTRIBUTE, messageAttributeValue); } Aws::SNS::Model::PublishOutcome outcome = snsClient.Publish(request); if (outcome.IsSuccess()) { std::cout << "Your message was successfully published." << std::endl; } else { std::cerr << "Error with TopicsAndQueues::Publish. " << outcome.GetError().GetMessage() << std::endl; cleanUp(topicARN, queueURLS, subscriptionARNS, snsClient, sqsClient); return false; } ``` + 有关 API 详细信息，请参阅《适用于 C\$1\$1 的 AWS SDK API Reference》**中的 [Publish](https://docs.aws.amazon.com/goto/SdkForCpp/sns-2010-03-31/Publish)。 ------ #### [ CLI ] **AWS CLI** **示例 1：向主题发布消息** 以下 `publish` 示例将指定消息发布到指定 SNS 主题。该消息来自一个文本文件，您可以在该文件中包含换行符。 ``` aws sns publish \ --topic-arn "arn:aws:sns:us-west-2:123456789012:my-topic" \ --message file://message.txt ``` `message.txt` 的内容： ``` Hello World Second Line ``` 输出： ``` { "MessageId": "123a45b6-7890-12c3-45d6-111122223333" } ``` **示例 2：向电话号码发布 SMS 消息** 以下 `publish` 示例将消息 `Hello world!` 发布到电话号码 `+1-555-555-0100`。 ``` aws sns publish \ --message "Hello world!" \ --phone-number +1-555-555-0100 ``` 输出： ``` { "MessageId": "123a45b6-7890-12c3-45d6-333322221111" } ``` + 有关 API 详细信息，请参阅《AWS CLI Command Reference》**中的 [Publish](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/sns/publish.html)。 ------ #### [ Go ] **适用于 Go 的 SDK V2** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/gov2/workflows/topics_and_queues#code-examples)中查找完整示例，了解如何进行设置和运行。 ``` import ( "context" "encoding/json" "log" "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/service/sns" "github.com/aws/aws-sdk-go-v2/service/sns/types" ) // SnsActions encapsulates the Amazon Simple Notification Service (Amazon SNS) actions // used in the examples. type SnsActions struct { SnsClient *sns.Client } // Publish publishes a message to an Amazon SNS topic. The message is then sent to all // subscribers. When the topic is a FIFO topic, the message must also contain a group ID // and, when ID-based deduplication is used, a deduplication ID. An optional key-value // filter attribute can be specified so that the message can be filtered according to // a filter policy. func (actor SnsActions) Publish(ctx context.Context, topicArn string, message string, groupId string, dedupId string, filterKey string, filterValue string) error { publishInput := sns.PublishInput{TopicArn: aws.String(topicArn), Message: aws.String(message)} if groupId != "" { publishInput.MessageGroupId = aws.String(groupId) } if dedupId != "" { publishInput.MessageDeduplicationId = aws.String(dedupId) } if filterKey != "" && filterValue != "" { publishInput.MessageAttributes = map[string]types.MessageAttributeValue{ filterKey: {DataType: aws.String("String"), StringValue: aws.String(filterValue)}, } } _, err := actor.SnsClient.Publish(ctx, &publishInput) if err != nil { log.Printf("Couldn't publish message to topic %v. Here's why: %v", topicArn, err) } return err } ``` + 有关 API 详细信息，请参阅《适用于 Go 的 AWS SDK API Reference》**中的 [Publish](https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/service/sns#Client.Publish)。 ------ #### [ Java ] **适用于 Java 的 SDK 2.x** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/example_code/sns#code-examples)中查找完整示例，了解如何进行设置和运行。 ``` import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.sns.SnsClient; import software.amazon.awssdk.services.sns.model.PublishRequest; import software.amazon.awssdk.services.sns.model.PublishResponse; import software.amazon.awssdk.services.sns.model.SnsException; /** * Before running this Java V2 code example, set up your development * environment, including your credentials. * * For more information, see the following documentation topic: * * https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/get-started.html */ public class PublishTopic { public static void main(String[] args) { final String usage = """ Usage: Where: message - The message text to send. topicArn - The ARN of the topic to publish. """; if (args.length != 2) { System.out.println(usage); System.exit(1); } String message = args[0]; String topicArn = args[1]; SnsClient snsClient = SnsClient.builder() .region(Region.US_EAST_1) .build(); pubTopic(snsClient, message, topicArn); snsClient.close(); } public static void pubTopic(SnsClient snsClient, String message, String topicArn) { try { PublishRequest request = PublishRequest.builder() .message(message) .topicArn(topicArn) .build(); PublishResponse result = snsClient.publish(request); System.out .println(result.messageId() + " Message sent. Status is " + result.sdkHttpResponse().statusCode()); } catch (SnsException e) { System.err.println(e.awsErrorDetails().errorMessage()); System.exit(1); } } } ``` + 有关 API 详细信息，请参阅《AWS SDK for Java 2.x API Reference》**中的 [Publish](https://docs.aws.amazon.com/goto/SdkForJavaV2/sns-2010-03-31/Publish)。 ------ #### [ JavaScript ] **适用于 JavaScript (v3) 的软件开发工具包** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javascriptv3/example_code/sns#code-examples) 中查找完整示例，了解如何进行设置和运行。在单独的模块中创建客户端并将其导出。 ``` import { SNSClient } from "@aws-sdk/client-sns"; // The AWS Region can be provided here using the `region` property. If you leave it blank // the SDK will default to the region set in your AWS config. export const snsClient = new SNSClient({}); ``` 导入 SDK 和客户端模块，然后调用 API。 ``` import { PublishCommand } from "@aws-sdk/client-sns"; import { snsClient } from "../libs/snsClient.js"; /** * @param {string | Record} message - The message to send. Can be a plain string or an object * if you are using the `json` `MessageStructure`. * @param {string} topicArn - The ARN of the topic to which you would like to publish. */ export const publish = async ( message = "Hello from SNS!", topicArn = "TOPIC_ARN", ) => { const response = await snsClient.send( new PublishCommand({ Message: message, TopicArn: topicArn, }), ); console.log(response); // { // '$metadata': { // httpStatusCode: 200, // requestId: 'e7f77526-e295-5325-9ee4-281a43ad1f05', // extendedRequestId: undefined, // cfId: undefined, // attempts: 1, // totalRetryDelay: 0 // }, // MessageId: 'xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx' // } return response; }; ``` 使用组、复制和属性选项向主题发布消息。 ``` async publishMessages() { const message = await this.prompter.input({ message: MESSAGES.publishMessagePrompt, }); let groupId; let deduplicationId; let choices; if (this.isFifo) { await this.logger.log(MESSAGES.groupIdNotice); groupId = await this.prompter.input({ message: MESSAGES.groupIdPrompt, }); if (this.autoDedup === false) { await this.logger.log(MESSAGES.deduplicationIdNotice); deduplicationId = await this.prompter.input({ message: MESSAGES.deduplicationIdPrompt, }); } choices = await this.prompter.checkbox({ message: MESSAGES.messageAttributesPrompt, choices: toneChoices, }); } await this.snsClient.send( new PublishCommand({ TopicArn: this.topicArn, Message: message, ...(groupId ? { MessageGroupId: groupId, } : {}), ...(deduplicationId ? { MessageDeduplicationId: deduplicationId, } : {}), ...(choices ? { MessageAttributes: { tone: { DataType: "String.Array", StringValue: JSON.stringify(choices), }, }, } : {}), }), ); const publishAnother = await this.prompter.confirm({ message: MESSAGES.publishAnother, }); if (publishAnother) { await this.publishMessages(); } } ``` + 有关更多信息，请参阅《适用于 JavaScript 的 AWS SDK 开发人员指南》[https://docs.aws.amazon.com/sdk-for-javascript/v3/developer-guide/sns-examples-publishing-messages.html](https://docs.aws.amazon.com/sdk-for-javascript/v3/developer-guide/sns-examples-publishing-messages.html)。 + 有关 API 详细信息，请参阅 *适用于 JavaScript 的 AWS SDK API 参考*中的 [Publish](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/sns/command/PublishCommand)。 ------ #### [ Kotlin ] **适用于 Kotlin 的 SDK** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/kotlin/services/sns#code-examples)中查找完整示例，了解如何进行设置和运行。 ``` suspend fun pubTopic( topicArnVal: String, messageVal: String, ) { val request = PublishRequest { message = messageVal topicArn = topicArnVal } SnsClient.fromEnvironment { region = "us-east-1" }.use { snsClient -> val result = snsClient.publish(request) println("${result.messageId} message sent.") } } ``` + 有关 API 详细信息，请参阅《AWS SDK for Kotlin API Reference》**中的 [Publish](https://sdk.amazonaws.com/kotlin/api/latest/index.html)。 ------ #### [ PHP ] **适用于 PHP 的 SDK** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/php/example_code/sns#code-examples)中查找完整示例，了解如何进行设置和运行。 ``` require 'vendor/autoload.php'; use Aws\Exception\AwsException; use Aws\Sns\SnsClient; /** * Sends a message to an Amazon SNS topic. * * This code expects that you have AWS credentials set up per: * https://docs.aws.amazon.com/sdk-for-php/v3/developer-guide/guide_credentials.html */ $SnSclient = new SnsClient([ 'profile' => 'default', 'region' => 'us-east-1', 'version' => '2010-03-31' ]); $message = 'This message is sent from a Amazon SNS code sample.'; $topic = 'arn:aws:sns:us-east-1:111122223333:MyTopic'; try { $result = $SnSclient->publish([ 'Message' => $message, 'TopicArn' => $topic, ]); var_dump($result); } catch (AwsException $e) { // output error message if fails error_log($e->getMessage()); } ``` + 有关更多信息，请参阅《适用于 PHP 的 AWS SDK 开发人员指南》[https://docs.aws.amazon.com/sdk-for-php/v3/developer-guide/sns-examples-subscribing-unsubscribing-topics.html#publish-a-message-to-an-sns-topic](https://docs.aws.amazon.com/sdk-for-php/v3/developer-guide/sns-examples-subscribing-unsubscribing-topics.html#publish-a-message-to-an-sns-topic)。 + 有关 API 的详细信息，请参阅 *适用于 PHP 的 AWS SDK API 参考*中的 [Publish](https://docs.aws.amazon.com/goto/SdkForPHPV3/sns-2010-03-31/Publish)。 ------ #### [ PowerShell ] **适用于 PowerShell V4 的工具** **示例 1：此示例显示发布一条 MessageAttribute 声明为内联的消息。** ``` Publish-SNSMessage -TopicArn "arn:aws:sns:us-west-2:123456789012:my-topic" -Message "Hello" -MessageAttribute @{'City'=[Amazon.SimpleNotificationService.Model.MessageAttributeValue]@{DataType='String'; StringValue ='AnyCity'}} ``` **示例 2：此示例显示发布一条事先 MessageAttributes 声明了多个消息的情况。** ``` $cityAttributeValue = New-Object Amazon.SimpleNotificationService.Model.MessageAttributeValue $cityAttributeValue.DataType = "String" $cityAttributeValue.StringValue = "AnyCity" $populationAttributeValue = New-Object Amazon.SimpleNotificationService.Model.MessageAttributeValue $populationAttributeValue.DataType = "Number" $populationAttributeValue.StringValue = "1250800" $messageAttributes = New-Object System.Collections.Hashtable $messageAttributes.Add("City", $cityAttributeValue) $messageAttributes.Add("Population", $populationAttributeValue) Publish-SNSMessage -TopicArn "arn:aws:sns:us-west-2:123456789012:my-topic" -Message "Hello" -MessageAttribute $messageAttributes ``` + 有关 API 详细信息，请参阅《AWS Tools for PowerShell Cmdlet Reference (V4)》**中的 [Publish](https://docs.aws.amazon.com/powershell/v4/reference)。 **适用于 PowerShell V5 的工具** **示例 1：此示例显示发布一条 MessageAttribute 声明为内联的消息。** ``` Publish-SNSMessage -TopicArn "arn:aws:sns:us-west-2:123456789012:my-topic" -Message "Hello" -MessageAttribute @{'City'=[Amazon.SimpleNotificationService.Model.MessageAttributeValue]@{DataType='String'; StringValue ='AnyCity'}} ``` **示例 2：此示例显示发布一条事先 MessageAttributes 声明了多个消息的情况。** ``` $cityAttributeValue = New-Object Amazon.SimpleNotificationService.Model.MessageAttributeValue $cityAttributeValue.DataType = "String" $cityAttributeValue.StringValue = "AnyCity" $populationAttributeValue = New-Object Amazon.SimpleNotificationService.Model.MessageAttributeValue $populationAttributeValue.DataType = "Number" $populationAttributeValue.StringValue = "1250800" $messageAttributes = New-Object System.Collections.Hashtable $messageAttributes.Add("City", $cityAttributeValue) $messageAttributes.Add("Population", $populationAttributeValue) Publish-SNSMessage -TopicArn "arn:aws:sns:us-west-2:123456789012:my-topic" -Message "Hello" -MessageAttribute $messageAttributes ``` + 有关 API 详细信息，请参阅《AWS Tools for PowerShell Cmdlet Reference (V5)》**中的 [Publish](https://docs.aws.amazon.com/powershell/v5/reference)。 ------ #### [ Python ] **适用于 Python 的 SDK（Boto3）** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/sns#code-examples)中查找完整示例，了解如何进行设置和运行。发布包含属性的消息，以便订阅可以根据属性进行筛选。 ``` class SnsWrapper: """Encapsulates Amazon SNS topic and subscription functions.""" def __init__(self, sns_resource): """ :param sns_resource: A Boto3 Amazon SNS resource. """ self.sns_resource = sns_resource @staticmethod def publish_message(topic, message, attributes): """ Publishes a message, with attributes, to a topic. Subscriptions can be filtered based on message attributes so that a subscription receives messages only when specified attributes are present. :param topic: The topic to publish to. :param message: The message to publish. :param attributes: The key-value attributes to attach to the message. Values must be either `str` or `bytes`. :return: The ID of the message. """ try: att_dict = {} for key, value in attributes.items(): if isinstance(value, str): att_dict[key] = {"DataType": "String", "StringValue": value} elif isinstance(value, bytes): att_dict[key] = {"DataType": "Binary", "BinaryValue": value} response = topic.publish(Message=message, MessageAttributes=att_dict) message_id = response["MessageId"] logger.info( "Published message with attributes %s to topic %s.", attributes, topic.arn, ) except ClientError: logger.exception("Couldn't publish message to topic %s.", topic.arn) raise else: return message_id ``` 发布基于订阅用户的协议采取不同形式的消息。 ``` class SnsWrapper: """Encapsulates Amazon SNS topic and subscription functions.""" def __init__(self, sns_resource): """ :param sns_resource: A Boto3 Amazon SNS resource. """ self.sns_resource = sns_resource @staticmethod def publish_multi_message( topic, subject, default_message, sms_message, email_message ): """ Publishes a multi-format message to a topic. A multi-format message takes different forms based on the protocol of the subscriber. For example, an SMS subscriber might receive a short version of the message while an email subscriber could receive a longer version. :param topic: The topic to publish to. :param subject: The subject of the message. :param default_message: The default version of the message. This version is sent to subscribers that have protocols that are not otherwise specified in the structured message. :param sms_message: The version of the message sent to SMS subscribers. :param email_message: The version of the message sent to email subscribers. :return: The ID of the message. """ try: message = { "default": default_message, "sms": sms_message, "email": email_message, } response = topic.publish( Message=json.dumps(message), Subject=subject, MessageStructure="json" ) message_id = response["MessageId"] logger.info("Published multi-format message to topic %s.", topic.arn) except ClientError: logger.exception("Couldn't publish message to topic %s.", topic.arn) raise else: return message_id ``` ``` class SnsWrapper: """Wrapper class for managing Amazon SNS operations.""" def __init__(self, sns_client: Any) -> None: """ Initialize the SnsWrapper. :param sns_client: A Boto3 Amazon SNS client. """ self.sns_client = sns_client @classmethod def from_client(cls) -> 'SnsWrapper': """ Create an SnsWrapper instance using a default boto3 client. :return: An instance of this class. """ sns_client = boto3.client('sns') return cls(sns_client) def publish_message( self, topic_arn: str, message: str, tone_attribute: Optional[str] = None, deduplication_id: Optional[str] = None, message_group_id: Optional[str] = None ) -> str: """ Publish a message to an SNS topic. :param topic_arn: The ARN of the SNS topic. :param message: The message content to publish. :param tone_attribute: Optional tone attribute for message filtering. :param deduplication_id: Optional deduplication ID for FIFO topics. :param message_group_id: Optional message group ID for FIFO topics. :return: The message ID of the published message. :raises ClientError: If the message publication fails. """ try: publish_args = { 'TopicArn': topic_arn, 'Message': message } # Add message attributes if tone is specified if tone_attribute: publish_args['MessageAttributes'] = { 'tone': { 'DataType': 'String', 'StringValue': tone_attribute } } # Add FIFO-specific parameters if message_group_id: publish_args['MessageGroupId'] = message_group_id if deduplication_id: publish_args['MessageDeduplicationId'] = deduplication_id response = self.sns_client.publish(**publish_args) message_id = response['MessageId'] logger.info(f"Published message to topic {topic_arn} with ID: {message_id}") return message_id except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') logger.error(f"Error publishing message to topic: {error_code} - {e}") raise ``` + 有关 API 详细信息，请参阅《AWS SDK for Python (Boto3) API Reference》**中的 [Publish](https://docs.aws.amazon.com/goto/boto3/sns-2010-03-31/Publish)。 ------ #### [ Ruby ] **适用于 Ruby 的 SDK** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/ruby/example_code/sns#code-examples)中查找完整示例，了解如何进行设置和运行。 ``` # Service class for sending messages using Amazon Simple Notification Service (SNS) class SnsMessageSender # Initializes the SnsMessageSender with an SNS client # # @param sns_client [Aws::SNS::Client] The SNS client def initialize(sns_client) @sns_client = sns_client @logger = Logger.new($stdout) end # Sends a message to a specified SNS topic # # @param topic_arn [String] The ARN of the SNS topic # @param message [String] The message to send # @return [Boolean] true if message was successfully sent, false otherwise def send_message(topic_arn, message) @sns_client.publish(topic_arn: topic_arn, message: message) @logger.info("Message sent successfully to #{topic_arn}.") true rescue Aws::SNS::Errors::ServiceError => e @logger.error("Error while sending the message: #{e.message}") false end end # Example usage: if $PROGRAM_NAME == __FILE__ topic_arn = 'SNS_TOPIC_ARN' # Should be replaced with a real topic ARN message = 'MESSAGE' # Should be replaced with the actual message content sns_client = Aws::SNS::Client.new message_sender = SnsMessageSender.new(sns_client) @logger.info('Sending message.') unless message_sender.send_message(topic_arn, message) @logger.error('Message sending failed. Stopping program.') exit 1 end end ``` + 有关更多信息，请参阅《适用于 Ruby 的 AWS SDK 开发人员指南》[https://docs.aws.amazon.com/sdk-for-ruby/v3/developer-guide/sns-example-send-message.html](https://docs.aws.amazon.com/sdk-for-ruby/v3/developer-guide/sns-example-send-message.html)。 + 有关 API 详细信息，请参阅 *适用于 Ruby 的 AWS SDK API 参考*中的 [Publish](https://docs.aws.amazon.com/goto/SdkForRubyV3/sns-2010-03-31/Publish)。 ------ #### [ Rust ] **SDK for Rust** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/rustv1/examples/sns#code-examples)中查找完整示例，了解如何进行设置和运行。 ``` async fn subscribe_and_publish( client: &Client, topic_arn: &str, email_address: &str, ) -> Result<(), Error> { println!("Receiving on topic with ARN: `{}`", topic_arn); let rsp = client .subscribe() .topic_arn(topic_arn) .protocol("email") .endpoint(email_address) .send() .await?; println!("Added a subscription: {:?}", rsp); let rsp = client .publish() .topic_arn(topic_arn) .message("hello sns!") .send() .await?; println!("Published message: {:?}", rsp); Ok(()) } ``` + 有关 API 详细信息，请参阅《AWS SDK for Rust API Reference》**中的 [Publish](https://docs.rs/aws-sdk-sns/latest/aws_sdk_sns/client/struct.Client.html#method.publish)。 ------ #### [ SAP ABAP ] **适用于 SAP ABAP 的 SDK** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/sap-abap/services/sns#code-examples)中查找完整示例，了解如何进行设置和运行。 ``` TRY. oo_result = lo_sns->publish( " oo_result is returned for testing purposes. " iv_topicarn = iv_topic_arn iv_message = iv_message ). MESSAGE 'Message published to SNS topic.' TYPE 'I'. CATCH /aws1/cx_snsnotfoundexception. MESSAGE 'Topic does not exist.' TYPE 'E'. ENDTRY. ``` 发布带有主题属性的消息。 ``` TRY. oo_result = lo_sns->publish( " oo_result is returned for testing purposes. " iv_topicarn = iv_topic_arn iv_message = iv_message it_messageattributes = it_msg_attrs ). MESSAGE 'Message with attributes published to SNS topic.' TYPE 'I'. CATCH /aws1/cx_snsnotfoundexception. MESSAGE 'Topic does not exist.' TYPE 'E'. ENDTRY. ``` 向主题发布多格式消息。 ``` " Build JSON message structure for multi-format message DATA(lv_json_message) = |\{ "default": "{ iv_default_message }", "sms": "{ iv_sms_message }", "email": "{ iv_email_message }" \}|. TRY. oo_result = lo_sns->publish( " oo_result is returned for testing purposes. " iv_topicarn = iv_topic_arn iv_message = lv_json_message iv_subject = iv_subject iv_messagestructure = 'json' ). MESSAGE 'Multi-format message published to SNS topic.' TYPE 'I'. CATCH /aws1/cx_snsnotfoundexception. MESSAGE 'Topic does not exist.' TYPE 'E'. ENDTRY. ``` + 有关 API 详细信息，请参阅《AWS SDK for SAP ABAP API Reference》**中的 [Publish](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)。 ------ #### [ Swift ] **适用于 Swift 的 SDK** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/swift/example_code/sns#code-examples)中查找完整示例，了解如何进行设置和运行。 ``` import AWSSNS let config = try await SNSClient.SNSClientConfiguration(region: region) let snsClient = SNSClient(config: config) let output = try await snsClient.publish( input: PublishInput( message: message, topicArn: arn ) ) guard let messageId = output.messageId else { print("No message ID received from Amazon SNS.") return } print("Published message with ID \(messageId)") ``` + 有关 API 详细信息，请参阅 *AWS SDK for Swift API Reference* 中的 [Publish](https://sdk.amazonaws.com/swift/api/awssns/latest/documentation/awssns/snsclient/publish(input:))。 ------ # 通过 Amazon SNS 和 Amazon S3 发布大型消息要发布很大的 Amazon SNS 消息，您可以使用[适用于 Java 的 Amazon SNS 扩展型客户端库](https://github.com/awslabs/amazon-sns-java-extended-client-lib/)或[适用于 Python 的 Amazon SNS 扩展型客户端库](https://github.com/awslabs/amazon-sns-python-extended-client-lib)。对于大于当前最大值 256KB（最大为 2GB）的消息，这些库非常有用。这两个库将实际有效负载保存到 Amazon S3 桶，并将存储的 Amazon S3 对象的引用发布到 Amazon SNS 主题。订阅的 Amazon SQS 队列可以使用[适用于 Java 的 Amazon SQS 扩展客户端库](https://github.com/awslabs/amazon-sqs-java-extended-client-lib)从 Amazon S3 中取消引用并检索负载。其他端点（如 Lambda）可以使用 [AWS的有效负载卸载 Java 公共库](https://github.com/awslabs/payload-offloading-java-common-lib-for-aws)来取消引用并检索有效负载。 **注意** Amazon SNS 扩展型客户端库与标准主题和 FIFO 主题都兼容。 # 适用于 Java 的 Amazon SNS 扩展型客户端库 ## 先决条件以下是使用[适用于 Java 的 Amazon SNS 扩展型客户端库](https://github.com/awslabs/amazon-sns-java-extended-client-lib)的先决条件： + 一个 AWS 软件开发工具包。本页上的示例使用 AWS Java 开发工具包。要安装和设置 SDK，请参阅《*适用于 Java 的 AWS SDK 开发者指南*[》中的 “设置 AWS 适用于 Java 的 SDK](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/setup-install.html)”。 + 并 AWS 账户具有正确的凭据。要创建 AWS 账户，请导航到[AWS 主页](https://aws.amazon.com/)，然后选择**创建 AWS 帐户**。按照说明进行操作。有关证书的信息，请参阅《*适用于 Java 的 AWS SDK 开发人员指南》中的 “设置用于开发*[的 AWS 凭证和区域](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/setup-credentials.html)”。 + Java 8 或更高版本。 + 适用于 Java 的 Amazon SNS 扩展型客户端库（也可从 [Maven](https://maven.apache.org/) 中获得）。 ## 配置消息存储 Amazon SNS 扩展客户端库使用负载卸载 Java 公共库 AWS 进行消息存储和检索。您可以配置以下 Amazon S3 [消息存储选项](https://github.com/awslabs/amazon-sns-java-extended-client-lib/blob/main/src/main/java/software/amazon/sns/SNSExtendedClientConfiguration.java)： + **自定义消息大小阈值** – 具有超过此大小的有效载荷和属性的消息将自动存储在 Amazon S3 中。 + **`alwaysThroughS3` 标志** – 将此值设置为 `true` 以强制将所有消息有效载荷存储在 Amazon S3 中。例如： ``` SNSExtendedClientConfiguration snsExtendedClientConfiguration = new SNSExtendedClientConfiguration() .withPayloadSupportEnabled(s3Client, BUCKET_NAME).withAlwaysThroughS3(true); ``` + **自定义 KMS 密钥** – 用于 Amazon S3 存储桶中的服务器端加密的密钥。 + **存储桶名称** – 用于存储消息有效载荷的 Amazon S3 存储桶的名称。 ## 示例：使用存储在 Amazon S3 中的负载将消息发布到 Amazon SNS 以下代码示例展示了如何： + 创建示例主题和队列。 + 订阅队列以接收来自主题的消息。 + 发布测试消息。消息负载存储在 Amazon S3，以及发布到的引用中。Amazon SQS 扩展客户端用于接收消息。 **适用于 Java 的 SDK 1.x** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/java/example_code/sns#code-examples)中查找完整示例，了解如何进行设置和运行。要发布大型消息，请使用适用于 Java 的 Amazon SNS 扩展客户端库。您发送的消息将引用包含实际消息内容的 Amazon S3 对象。 ``` import com.amazon.sqs.javamessaging.AmazonSQSExtendedClient; import com.amazon.sqs.javamessaging.ExtendedClientConfiguration; import com.amazonaws.regions.Region; import com.amazonaws.regions.Regions; import com.amazonaws.services.s3.AmazonS3; import com.amazonaws.services.s3.AmazonS3ClientBuilder; import com.amazonaws.services.sns.AmazonSNS; import com.amazonaws.services.sns.AmazonSNSClientBuilder; import com.amazonaws.services.sns.model.CreateTopicRequest; import com.amazonaws.services.sns.model.PublishRequest; import com.amazonaws.services.sns.model.SetSubscriptionAttributesRequest; import com.amazonaws.services.sns.util.Topics; import com.amazonaws.services.sqs.AmazonSQS; import com.amazonaws.services.sqs.AmazonSQSClientBuilder; import com.amazonaws.services.sqs.model.CreateQueueRequest; import com.amazonaws.services.sqs.model.ReceiveMessageResult; import software.amazon.sns.AmazonSNSExtendedClient; import software.amazon.sns.SNSExtendedClientConfiguration; public class Example { public static void main(String[] args) { final String BUCKET_NAME = "extended-client-bucket"; final String TOPIC_NAME = "extended-client-topic"; final String QUEUE_NAME = "extended-client-queue"; final Regions region = Regions.DEFAULT_REGION; // Message threshold controls the maximum message size that will be allowed to // be published // through SNS using the extended client. Payload of messages exceeding this // value will be stored in // S3. The default value of this parameter is 256 KB which is the maximum // message size in SNS (and SQS). final int EXTENDED_STORAGE_MESSAGE_SIZE_THRESHOLD = 32; // Initialize SNS, SQS and S3 clients final AmazonSNS snsClient = AmazonSNSClientBuilder.standard().withRegion(region).build(); final AmazonSQS sqsClient = AmazonSQSClientBuilder.standard().withRegion(region).build(); final AmazonS3 s3Client = AmazonS3ClientBuilder.standard().withRegion(region).build(); // Create bucket, topic, queue and subscription s3Client.createBucket(BUCKET_NAME); final String topicArn = snsClient.createTopic( new CreateTopicRequest().withName(TOPIC_NAME)).getTopicArn(); final String queueUrl = sqsClient.createQueue( new CreateQueueRequest().withQueueName(QUEUE_NAME)).getQueueUrl(); final String subscriptionArn = Topics.subscribeQueue( snsClient, sqsClient, topicArn, queueUrl); // To read message content stored in S3 transparently through SQS extended // client, // set the RawMessageDelivery subscription attribute to TRUE final SetSubscriptionAttributesRequest subscriptionAttributesRequest = new SetSubscriptionAttributesRequest(); subscriptionAttributesRequest.setSubscriptionArn(subscriptionArn); subscriptionAttributesRequest.setAttributeName("RawMessageDelivery"); subscriptionAttributesRequest.setAttributeValue("TRUE"); snsClient.setSubscriptionAttributes(subscriptionAttributesRequest); // Initialize SNS extended client // PayloadSizeThreshold triggers message content storage in S3 when the // threshold is exceeded // To store all messages content in S3, use AlwaysThroughS3 flag final SNSExtendedClientConfiguration snsExtendedClientConfiguration = new SNSExtendedClientConfiguration() .withPayloadSupportEnabled(s3Client, BUCKET_NAME) .withPayloadSizeThreshold(EXTENDED_STORAGE_MESSAGE_SIZE_THRESHOLD); final AmazonSNSExtendedClient snsExtendedClient = new AmazonSNSExtendedClient(snsClient, snsExtendedClientConfiguration); // Publish message via SNS with storage in S3 final String message = "This message is stored in S3 as it exceeds the threshold of 32 bytes set above."; snsExtendedClient.publish(topicArn, message); // Initialize SQS extended client final ExtendedClientConfiguration sqsExtendedClientConfiguration = new ExtendedClientConfiguration() .withPayloadSupportEnabled(s3Client, BUCKET_NAME); final AmazonSQSExtendedClient sqsExtendedClient = new AmazonSQSExtendedClient(sqsClient, sqsExtendedClientConfiguration); // Read the message from the queue final ReceiveMessageResult result = sqsExtendedClient.receiveMessage(queueUrl); System.out.println("Received message is " + result.getMessages().get(0).getBody()); } } ``` ## 其他终端节点协议 Amazon SNS 和 Amazon SQS 库都使用 [AWS的负载卸载 Java 公共库](https://github.com/awslabs/payload-offloading-java-common-lib-for-aws)通过 Amazon S3 存储和检索消息负载。任何启用 Java 的终端节点（例如，在 Java 中实施的 HTTPS 终端节点）都可以使用相同的库来取消引用消息内容。无法使用有效负载卸载 Java 公共库的终端节点仍然 AWS 可以发布存储在 Amazon S3 中的有效负载的消息。以下是由上面的代码示例发布的 Amazon S3 引用的示例： ``` [ "software.amazon.payloadoffloading.PayloadS3Pointer", { "s3BucketName": "extended-client-bucket", "s3Key": "xxxx-xxxxx-xxxxx-xxxxxx" } ] ``` # 适用于 Python 的 Amazon SNS 扩展型客户端库 ## 先决条件以下是使用[适用于 Python 的 Amazon SNS 扩展型客户端库](https://github.com/awslabs/amazon-sns-python-extended-client-lib)的先决条件： + 一个 AWS 软件开发工具包。本页上的示例使用 AWS Python SDK Boto3。要安装和设置 SDK，请参阅 [https://boto3.amazonaws.com/v1/documentation/api/latest/guide/quickstart.html](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/quickstart.html) 文档。 + 并 AWS 账户具有正确的凭据。要创建 AWS 账户，请导航到[AWS 主页](https://aws.amazon.com/)，然后选择**创建 AWS 帐户**。按照说明进行操作。有关凭证的信息，请参阅《AWS SDK for Python 开发人员指南》**中的[凭证](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/credentials.html)。 + Python 3.x（或更高版本）和 pip。 + 适用于 Python 的 Amazon SNS 扩展型客户端库（也可从 [PyPI](https://pypi.org/project/amazon-sns-extended-client/) 中获得）。 ## 配置消息存储 Boto3 Amazon [SN](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/sns.html#client) S 客户端[、主题[PlatformEndpoint](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/sns/platformendpoint/index.html)和对象上提供了以下属性，](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/sns/topic/index.html)用于配置 Amazon S3 消息存储选项。 + **`large_payload_support`** – 用于存储大型消息的 Amazon S3 存储桶名称。 + **`use_legacy_attribute`** – 如果设置为 `True`，则所有已发布的消息都使用旧版保留消息属性（`SQSLargePayloadSize`），而不是当前的保留消息属性（`ExtendedPayloadSize`）。 + **`message_size_threshold`** – 在大型消息桶中存储消息的阈值。不能低于 `0` 或超过 `262144`。默认值为 `262144`。 + **`always_through_s3`** – 如果为 `True`，则所有消息都存储在 Amazon S3 中。默认值为 `False`。 + **`s3_client`** – 用于在 Amazon S3 中存储对象的 Boto3 Amazon S3 `client` 对象。如果您想要控制 Amazon S3 客户端（例如自定义 Amazon S3 配置或凭证），请使用此选项。首次使用时，若未预先设置，默认值为 `boto3.client("s3")`。 ## 示例：使用存储在 Amazon S3 中的有效负载将消息发布到 Amazon SNS 以下代码示例展示了如何： + 创建示例 Amazon SNS 主题和 Amazon SQS 队列。 + 为 Amazon SQS 队列附加策略以接收来自 Amazon SNS 主题的消息。 + 订阅队列以接收来自主题的消息。 + 使用 Amazon SNS 扩展客户端、主题资源和 PlatformEndpoint 资源发布测试消息。 + 消息有效负载存储在 Amazon S3 中，并发布对它的引用。 + 打印队列中已发布的消息以及从 Amazon S3 检索到的原始消息。要发布大型消息，请使用适用于 Python 的 Amazon SNS 扩展型客户端库。您发送的消息将引用包含实际消息内容的 Amazon S3 对象。 ``` import boto3 from sns_extended_client import SNSExtendedClientSession from json import loads s3_extended_payload_bucket = "extended-client-bucket-store" # S3 bucket with the given bucket name is a resource which is created and accessible with the given AWS credentials TOPIC_NAME = "---TOPIC-NAME---" QUEUE_NAME = "---QUEUE-NAME---" def allow_sns_to_write_to_sqs(topicarn, queuearn): policy_document = """{{ "Version": "2012-10-17", "Statement":[ {{ "Sid":"MyPolicy", "Effect":"Allow", "Principal" : {{"AWS" : "*"}}, "Action":"SQS:SendMessage", "Resource": "{}", "Condition":{{ "ArnEquals":{{ "aws:SourceArn": "{}" }} }} }} ] }}""".format(queuearn, topicarn) return policy_document def get_msg_from_s3(body,sns_extended_client): """Handy Helper to fetch message from S3""" json_msg = loads(body) s3_object = sns_extended_client.s3_client.get_object( Bucket=json_msg[1].get("s3BucketName"), Key=json_msg[1].get("s3Key") ) msg = s3_object.get("Body").read().decode() return msg def fetch_and_print_from_sqs(sqs, queue_url,sns_extended_client): sqs_msg = sqs.receive_message( QueueUrl=queue_url, AttributeNames=['All'], MessageAttributeNames=['All'], VisibilityTimeout=0, WaitTimeSeconds=0, MaxNumberOfMessages=1 ).get("Messages")[0] message_body = sqs_msg.get("Body") print("Published Message: {}".format(message_body)) print("Message Stored in S3 Bucket is: {}\n".format(get_msg_from_s3(message_body,sns_extended_client))) # Delete the Processed Message sqs.delete_message( QueueUrl=queue_url, ReceiptHandle=sqs_msg['ReceiptHandle'] ) sns_extended_client = boto3.client("sns", region_name="us-east-1") create_topic_response = sns_extended_client.create_topic(Name=TOPIC_NAME) sns_topic_arn = create_topic_response.get("TopicArn") # create and subscribe an sqs queue to the sns client sqs = boto3.client("sqs",region_name="us-east-1") demo_queue_url = sqs.create_queue(QueueName=QUEUE_NAME).get("QueueUrl") sqs_queue_arn = sqs.get_queue_attributes( QueueUrl=demo_queue_url, AttributeNames=["QueueArn"] )["Attributes"].get("QueueArn") # Adding policy to SQS queue such that SNS topic can send msg to SQS queue policy_json = allow_sns_to_write_to_sqs(sns_topic_arn, sqs_queue_arn) response = sqs.set_queue_attributes( QueueUrl = demo_queue_url, Attributes = { 'Policy' : policy_json } ) # Set the RawMessageDelivery subscription attribute to TRUE if you want to use # SQSExtendedClient to help with retrieving msg from S3 sns_extended_client.subscribe(TopicArn=sns_topic_arn, Protocol="sqs", Endpoint=sqs_queue_arn , Attributes={"RawMessageDelivery":"true"} ) sns_extended_client.large_payload_support = s3_extended_payload_bucket # Change default s3_client attribute of sns_extended_client to use 'us-east-1' region sns_extended_client.s3_client = boto3.client("s3", region_name="us-east-1") # Below is the example that all the messages will be sent to the S3 bucket sns_extended_client.always_through_s3 = True sns_extended_client.publish( TopicArn=sns_topic_arn, Message="This message should be published to S3" ) print("\n\nPublished using SNS extended client:") fetch_and_print_from_sqs(sqs, demo_queue_url,sns_extended_client) # Prints message stored in s3 # Below is the example that all the messages larger than 32 bytes will be sent to the S3 bucket print("\nUsing decreased message size threshold:") sns_extended_client.always_through_s3 = False sns_extended_client.message_size_threshold = 32 sns_extended_client.publish( TopicArn=sns_topic_arn, Message="This message should be published to S3 as it exceeds the limit of the 32 bytes", ) fetch_and_print_from_sqs(sqs, demo_queue_url,sns_extended_client) # Prints message stored in s3 # Below is the example to publish message using the SNS.Topic resource sns_extended_client_resource = SNSExtendedClientSession().resource( "sns", region_name="us-east-1" ) topic = sns_extended_client_resource.Topic(sns_topic_arn) topic.large_payload_support = s3_extended_payload_bucket # Change default s3_client attribute of topic to use 'us-east-1' region topic.s3_client = boto3.client("s3", region_name="us-east-1") topic.always_through_s3 = True # Can Set custom S3 Keys to be used to store objects in S3 topic.publish( Message="This message should be published to S3 using the topic resource", MessageAttributes={ "S3Key": { "DataType": "String", "StringValue": "347c11c4-a22c-42e4-a6a2-9b5af5b76587", } }, ) print("\nPublished using Topic Resource:") fetch_and_print_from_sqs(sqs, demo_queue_url,topic) # Below is the example to publish message using the SNS.PlatformEndpoint resource sns_extended_client_resource = SNSExtendedClientSession().resource( "sns", region_name="us-east-1" ) platform_endpoint = sns_extended_client_resource.PlatformEndpoint(sns_topic_arn) platform_endpoint.large_payload_support = s3_extended_payload_bucket # Change default s3_client attribute of platform_endpoint to use 'us-east-1' region platform_endpoint.s3_client = boto3.client("s3", region_name="us-east-1") platform_endpoint.always_through_s3 = True # Can Set custom S3 Keys to be used to store objects in S3 platform_endpoint.publish( Message="This message should be published to S3 using the PlatformEndpoint resource", MessageAttributes={ "S3Key": { "DataType": "String", "StringValue": "247c11c4-a22c-42e4-a6a2-9b5af5b76587", } }, ) print("\nPublished using PlatformEndpoint Resource:") fetch_and_print_from_sqs(sqs, demo_queue_url,platform_endpoint) ``` **输出** ``` Published using SNS extended client: Published Message: ["software.amazon.payloadoffloading.PayloadS3Pointer", {"s3BucketName": "extended-client-bucket-store", "s3Key": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"}] Message Stored in S3 Bucket is: This message should be published to S3 Using decreased message size threshold: Published Message: ["software.amazon.payloadoffloading.PayloadS3Pointer", {"s3BucketName": "extended-client-bucket-store", "s3Key": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"}] Message Stored in S3 Bucket is: This message should be published to S3 as it exceeds the limit of the 32 bytes Published using Topic Resource: Published Message: ["software.amazon.payloadoffloading.PayloadS3Pointer", {"s3BucketName": "extended-client-bucket-store", "s3Key": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"}] Message Stored in S3 Bucket is: This message should be published to S3 using the topic resource Published using PlatformEndpoint Resource: Published Message: ["software.amazon.payloadoffloading.PayloadS3Pointer", {"s3BucketName": "extended-client-bucket-store", "s3Key": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"}] Message Stored in S3 Bucket is: This message should be published to S3 using the PlatformEndpoint resource ``` # Amazon SNS 消息属性 Amazon SNS 支持传输消息属性，这些属性支持您提供与消息相关的结构化元数据项目（如时间戳、地理空间数据、签名和标识符）。对于 SQS 订阅，在启用 [Raw Message Delivery](sns-large-payload-raw-message-delivery.md)（原始消息传输）时，最多可以发送 10 个消息属性。要发送 10 个以上的消息属性，必须禁用 Raw Message Delivery（原始消息传输）。所具有的消息属性（定向到启用了原始消息传输的 Amazon SQS 订阅）超过 10 个的消息将作为客户端错误被丢弃。消息属性是可选的，并独立于消息正文（但随之一起发送）。接收方可以使用此信息来决定如何处理消息，而不必先处理消息正文。有关使用 AWS 管理控制台或发送带有属性的消息的信息适用于 Java 的 AWS SDK，请参阅[要使用 Amazon SNS 主题发布消息 AWS 管理控制台](sns-publishing.md#sns-publishing-messages)教程。 **注意** 当消息结构为 String 而不是 JSON 时，仅发送消息属性。您还可以使用消息属性，帮助构造移动终端节点的推送通知消息。在这种情况下，消息属性仅用于帮助构造推送通知消息。这些属性不会被传递到终端节点，就像在发送带有消息属性的消息到 Amazon SQS 终端节点时一样。您还可以使用消息属性来让消息变为可通过订阅筛选策略进行筛选。可以将筛选策略应用于主题订阅。应用了筛选策略且筛选策略范围设置为 `MessageAttributes`（默认值）时，订阅将只接收具有策略接受的属性的那些消息。有关更多信息，请参阅 [Amazon SNS 消息筛选](sns-message-filtering.md)。 **注意** 使用消息属性进行筛选时，该值必须是有效的 JSON 字符串。这样做可以确保将消息传送到启用了消息属性筛选的订阅。 ## 消息属性项目和验证每个消息属性包含以下项目： + **Name** – 消息属性的名称可以包含以下字符：A-Z、a-z、0-9、下划线 (\$1)、连字符 (-) 和句点 (.)。名称不得以句点开头或结尾，并且不应包含连续句点。名称区分大小写，且必须在消息的所有属性名称中是唯一的。名称最多可以有 256 个字符。名称不能以 `AWS.` 或 `Amazon.` （或任何大小写变化形式）开头，因为这些前缀已预留以供 Amazon Web Services 使用。 + **Type** – 受支持的消息属性数据类型有 `String`、`String.Array`、`Number` 和 `Binary`。数据类型在内容方面具有与消息正文相同的限制。想要了解更多信息，请参阅[消息属性数据类型和验证](#SNSMessageAttributes.DataTypes)部分。 + **Value** – 用户指定的消息属性值。对于字符串数据类型，值属性的内容必须遵循与消息正文相同的内容限制。不过，如果消息属性用于过滤，则该值必须是有效的 JSON 字符串，以确保与 Amazon SNS 订阅筛选策略兼容。有关更多信息，请参阅 *Amazon Simple Notification Service API 参考*中的[发布](https://docs.aws.amazon.com/sns/latest/api/API_Publish.html)操作。名称、类型和值都不得为空或 null。此外，消息正文也不应为空或 null。消息属性的所有部分 (包括名称、类型和值) 都包含在消息大小限制中，该限制当前是 256 KB。 ## 消息属性数据类型和验证消息属性数据类型指示 Amazon SNS 处理消息属性值的方式。例如，如果类型是数字，则 Amazon SNS 会验证它是否为数字。除非另有说明，否则 Amazon SNS 支持所有终端节点的以下逻辑数据类型： + **String** – 字符串是使用 UTF-8 二进制编码的 Unicode。有关代码值的列表，请参阅 [http://en.wikipedia。 org/wiki/ASCII](http://en.wikipedia.org/wiki/ASCII#ASCII_printable_characters)\$1ASCII\$1printable\$1characters。 **注意** 消息属性中不支持代理值。例如，使用代理值来表示表情符号将收到以下错误：`Invalid attribute value was passed in for message attribute`。 + **String.Array** – 格式为字符串的阵列，可以包含多个值。这些值可以是字符串、数字或关键字 `true`、`false` 和 `null`。数字或布尔类型的 String.Array 不需要引号。多个 String.Array 值用逗号分隔。 AWS Lambda 订阅不支持此数据类型。如果您为 Lambda 终端节点指定此数据类型，它会作为 Amazon SNS 传输给 Lambda 的 JSON 负载中的 `String` 数据类型传递。 + **Number** – 数字是正或负整数或是浮点数。数字具有足够的范围和精度，以便包含整数、浮点数和双精度数通常支持的大多数可能值。数字的值可以介于 -109 到 109 之间，精确至小数点后 5 位数。系统会删减开头和结尾的 0。 AWS Lambda 订阅不支持此数据类型。如果您为 Lambda 终端节点指定此数据类型，它会作为 Amazon SNS 传输给 Lambda 的 JSON 负载中的 `String` 数据类型传递。 + **Binary** – 二进制类型属性可以存储任何二进制数据，例如压缩数据、加密数据或图像。 ## 为移动推送通知预留的消息属性下表列出了可用于构造推送通知消息的移动推送通知服务的预留消息属性： [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_cn/sns/latest/dg/sns-message-attributes.html) 1 如果消息属性不符合要求，Apple 将拒绝 Amazon SNS 通知。如需了解更多详情，请参阅 Apple 开发者网站上的 “[向发送通知请求](https://developer.apple.com/documentation/usernotifications/setting_up_a_remote_notification_server/sending_notification_requests_to_apns)”。 APNs # Amazon SNS 消息批处理 ## 什么是消息批处理？在各个 `Publish` API 请求中向标准或 FIFO 主题发布消息的替代方法，该方法使用 Amazon SNS `PublishBatch` API 在单个 API 请求中最多发布 10 条消息。批量发送消息可以帮助您利用 Amazon SNS 将与连接分布式应用程序（[A2A 消息收发](sns-system-to-system-messaging.md)）或者向人们发送通知（[A2P 消息收发](sns-user-notifications.md)）相关的成本最高降低 10 倍。根据您运营所在区域，Amazon SNS 对每秒可以向某个主题发布的消息数量设定了配额。有关 API 限额的更多信息，请参阅《AWS 一般参考》**中的 [Amazon SNS 端点和限额](https://docs.aws.amazon.com/general/latest/gr/sns.html)页面。 **注意** 您在单个 `PublishBatch` API 请求中发送的所有消息的总合计大小不能超过 262144 字节（256 KiB）。 `PublishBatch` API 将相同的 `Publish` API 操作用于 IAM 策略。 ## 消息批处理是如何工作的？使用 `PublishBatch` API 发布消息与使用 `Publish` API 发布消息类似。主要的区别在于，需要给 `PublishBatch` API 请求中的每条消息分配一个唯一的批处理 ID（最多 80 个字符）。这样，Amazon SNS 可以为批处理中的每条消息返回单独的 API 响应，以确认每条消息均已发布或发生故障。对于发布到 FIFO 主题的消息，除了包括分配唯一的批处理 ID 之外，还需要为每条单独的消息包括 `MessageDeduplicationID` 和 `MessageGroupId`。 ## 示例 **向标准主题发布一批 10 条的消息** ``` // Imports import software.amazon.awssdk.services.sns.SnsClient; import software.amazon.awssdk.services.sns.model.PublishBatchRequest; import software.amazon.awssdk.services.sns.model.PublishBatchRequestEntry; import software.amazon.awssdk.services.sns.model.PublishBatchResponse; import software.amazon.awssdk.services.sns.model.SnsException; import java.util.List; import java.util.stream.Collectors; import java.util.stream.IntStream; // Code private static final int MAX_BATCH_SIZE = 10; public static void publishBatchToTopic(SnsClient snsClient, String topicArn, int batchSize) { try { // Validate the batch size if (batchSize > MAX_BATCH_SIZE) { throw new IllegalArgumentException("Batch size cannot exceed " + MAX_BATCH_SIZE); } // Create the batch entries List entries = IntStream.range(0, batchSize) .mapToObj(i -> PublishBatchRequestEntry.builder() .id("id" + i) .message("message" + i) .build()) .collect(Collectors.toList()); // Build the batch request PublishBatchRequest request = PublishBatchRequest.builder() .topicArn(topicArn) .publishBatchRequestEntries(entries) .build(); // Publish the batch request PublishBatchResponse response = snsClient.publishBatch(request); // Handle successful messages response.successful().forEach(success -> { System.out.println("Successful Batch Id: " + success.id()); System.out.println("Message Id: " + success.messageId()); }); // Handle failed messages response.failed().forEach(failure -> { System.err.println("Failed Batch Id: " + failure.id()); System.err.println("Error Code: " + failure.code()); System.err.println("Sender Fault: " + failure.senderFault()); System.err.println("Error Message: " + failure.message()); }); } catch (SnsException e) { // Log and handle exceptions System.err.println("SNS Exception: " + e.awsErrorDetails().errorMessage()); } catch (IllegalArgumentException e) { System.err.println("Validation Error: " + e.getMessage()); } } ``` **向 FIFO 主题发布一批 10 条的消息** ``` // Imports import software.amazon.awssdk.services.sns.SnsClient; import software.amazon.awssdk.services.sns.model.PublishBatchRequest; import software.amazon.awssdk.services.sns.model.PublishBatchRequestEntry; import software.amazon.awssdk.services.sns.model.PublishBatchResponse; import software.amazon.awssdk.services.sns.model.BatchResultErrorEntry; import software.amazon.awssdk.services.sns.model.SnsException; import java.util.List; import java.util.stream.Collectors; import java.util.stream.IntStream; // Code private static final int MAX_BATCH_SIZE = 10; public static void publishBatchToFifoTopic(SnsClient snsClient, String topicArn) { try { // Create the batch entries to send List entries = IntStream.range(0, MAX_BATCH_SIZE) .mapToObj(i -> PublishBatchRequestEntry.builder() .id("id" + i) .message("message" + i) .messageGroupId("groupId") .messageDeduplicationId("deduplicationId" + i) .build()) .collect(Collectors.toList()); // Create the batch request PublishBatchRequest request = PublishBatchRequest.builder() .topicArn(topicArn) .publishBatchRequestEntries(entries) .build(); // Publish the batch request PublishBatchResponse response = snsClient.publishBatch(request); // Handle the successfully sent messages response.successful().forEach(success -> { System.out.println("Batch Id for successful message: " + success.id()); System.out.println("Message Id for successful message: " + success.messageId()); System.out.println("Sequence Number for successful message: " + success.sequenceNumber()); }); // Handle the failed messages response.failed().forEach(failure -> { System.err.println("Batch Id for failed message: " + failure.id()); System.err.println("Error Code for failed message: " + failure.code()); System.err.println("Sender Fault for failed message: " + failure.senderFault()); System.err.println("Failure Message for failed message: " + failure.message()); }); } catch (SnsException e) { // Handle any exceptions from the request System.err.println("SNS Exception: " + e.awsErrorDetails().errorMessage()); } } ```