本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。
# 使用 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)。這些程式庫對於大於目前最大 256 KB 的訊息 (最大 2 GB) 非常有用。程式庫會將實際酬載儲存到 Amazon S3 儲存貯體，並將儲存的 Amazon S3 物件的參考發布到該主題。訂閱的 Amazon SQS 佇列可以使用[適用於 Java 的 Amazon SQS 擴展客戶端程式庫](https://github.com/awslabs/amazon-sqs-java-extended-client-lib)從 Amazon S3 取消參照和擷取酬載資料。其他端點 (例如 Lambda) 可以使用[酬載卸載 Java 通用程式庫 AWS](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 開發人員指南*》中的[設定適用於 Java 的 AWS 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` flag** – 將此值設定為 `true`，強制將所有訊息承載儲存在 Amazon S3 中。例如：
```
SNSExtendedClientConfiguration snsExtendedClientConfiguration = new
SNSExtendedClientConfiguration() .withPayloadSupportEnabled(s3Client, BUCKET_NAME).withAlwaysThroughS3(true);
```
+ **自訂 KMS 金鑰** – 用於 Amazon S3 儲存貯體中伺服器端加密的金鑰。
+ **儲存貯體名稱** – 用於存放訊息承載的 Amazon S3 儲存貯體名稱。
## 範例：將訊息發布到 Amazon SNS，其中儲存在 Amazon S3 中的有效酬載
以下程式碼範例顯示做法：
+ 建立範例主題和佇列。
+ 訂閱佇列以接收來自主題的訊息。
+ 發布測試訊息。
訊息酬載儲存在 Amazon S3 中，並發布對該訊息的參考。Amazon SQS 延伸用戶端是用來接收訊息。
**適用於 Java 1.x 的 SDK**
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 擴充用戶端程式庫
## 先決條件
下列是使用[適用於 Java 的 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 SNS [用戶端](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/sns.html#client)、[主題](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/sns/topic/index.html)和[PlatformEndPoint](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/sns/platformendpoint/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`** – 用來將`client`物件存放至 Amazon S3 的 Boto3 Amazon S3 物件。如果您想要控制 Amazon S3 用戶端 （例如，自訂 Amazon S3 組態或登入資料），請使用此選項。如果先前未設定，則第一次使用`boto3.client("s3")`時預設為 。
## 範例：將訊息發布到 Amazon SNS，其中儲存在 Amazon S3 中的有效酬載
以下程式碼範例顯示做法：
+ 建立範例 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
```