适用于 Python 的 Amazon SNS 扩展型客户端库 - Amazon Simple Notification Service
先决条件配置消息存储示例：使用存储在 Amazon S3 中的有效负载将消息发布到 Amazon SNS

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

适用于 Python 的 Amazon SNS 扩展型客户端库

先决条件

以下是使用适用于 Python 的 Amazon SNS 扩展型客户端库的先决条件：

  • 一个 AWS 软件开发工具包。

    本页上的示例使用 AWS Python SDK Boto3。要安装和设置 SDK，请参阅 AWS SDK for Python 文档。

  • 并 AWS 账户 具有正确的凭据。

    要创建 AWS 账户，请导航到AWS 主页，然后选择创建 AWS 帐户。按照说明进行操作。

    有关凭证的信息，请参阅《AWS SDK for Python 开发人员指南》中的凭证。

  • Python 3.x（或更高版本）和 pip。

  • 适用于 Python 的 Amazon SNS 扩展型客户端库（也可从 PyPI 中获得）。

配置消息存储

Boto3 Amazon SN S 客户端、主题PlatformEndpoint和对象上提供了以下属性，用于配置 Amazon S3 消息存储选项。

  • large_payload_support – 用于存储大型消息的 Amazon S3 桶名称。

  • message_size_threshold – 在大型消息桶中存储消息的阈值。不能低于 0 或超过 262144。原定设置值为 262144。

  • always_through_s3 – 如果为 True，则所有消息都存储在 Amazon S3 中。默认为 False。

  • s3 – 用于在 Amazon S3 中存储对象的 Boto3 Amazon S3 resource 对象。如果您想要控制 Amazon S3 资源（例如，自定义 Amazon S3 配置或凭证），请使用此选项。如果之前在首次使用时未设置，则原定设置值为 boto3.resource("s3")。

示例：使用存储在 Amazon S3 中的有效负载将消息发布到 Amazon SNS

以下代码示例展示了如何：

  • 创建示例 Amazon SNS 主题和 Amazon SQS 队列。

  • 订阅队列以接收来自主题的消息。

  • 发布测试消息。

  • 消息有效负载存储在 Amazon S3 中，并发布对它的引用。

  • 打印队列中已发布的消息以及从 Amazon S3 检索到的原始消息。

要发布大型消息，请使用适用于 Python 的 Amazon SNS 扩展型客户端库。您发送的消息将引用包含实际消息内容的 Amazon S3 对象。

import boto3
import sns_extended_client
from json import loads

s3_extended_payload_bucket = "extended-client-bucket-store"  
TOPIC_NAME = "TOPIC-NAME"
QUEUE_NAME = "QUEUE-NAME"

# Create an helper to fetch message from S3 
def get_msg_from_s3(body):
    json_msg = loads(body)
    s3_client = boto3.client("s3")
    s3_object = 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

# Create an helper to fetch and print message SQS queue and S3        
def fetch_and_print_from_sqs(sqs, queue_url):
    """Handy Helper to fetch and print message from SQS queue and S3"""
    message = sqs.receive_message(
        QueueUrl=queue_url, MessageAttributeNames=["All"], MaxNumberOfMessages=1
    ).get("Messages")[0]
    message_body = message.get("Body")
    print("Published Message: {}".format(message_body))
    print("Message Stored in S3 Bucket is: {}\n".format(get_msg_from_s3(message_body)))

# Initialize the SNS client and create SNS Topic
sns_extended_client = boto3.client("sns", region_name="us-east-1")
create_topic_response = sns_extended_client.create_topic(Name=TOPIC_NAME)
demo_topic_arn = create_topic_response.get("TopicArn")

# Create and subscribe an SQS queue to the SNS client
sqs = boto3.client("sqs")
demo_queue_url = sqs.create_queue(QueueName=QUEUE_NAME).get("QueueUrl")
demo_queue_arn = sqs.get_queue_attributes(QueueUrl=demo_queue_url, 
AttributeNames=["QueueArn"])["Attributes"].get("QueueArn")
# Set the RawMessageDelivery subscription attribute to TRUE
sns_extended_client.subscribe(TopicArn=demo_topic_arn, Protocol="sqs", 
Endpoint=demo_queue_arn, Attributes={"RawMessageDelivery":"true"})

sns_extended_client.large_payload_support = s3_extended_payload_bucket

# To store all messages content in S3, set always_through_s3 to True
# In the example, we set message size threshold as 32 bytes, adjust this threshold as per your usecase
# Message will only be uploaded to S3 when its payload size exceeded threshold
sns_extended_client.message_size_threshold = 32
sns_extended_client.publish(
    TopicArn=demo_topic_arn,
    Message="This message should be published to S3 as it exceeds the message_size_threshold limit",
)
# Print message stored in s3
fetch_and_print_from_sqs(sqs, demo_queue_url)

输出

Published Message:
[
    "software.amazon.payloadoffloading.PayloadS3Pointer",
    {
        "s3BucketName": "extended-client-bucket-store",
        "s3Key": "xxxx-xxxxx-xxxxx-xxxxxx"
    }
]
Message Stored in S3 Bucket is: This message should be published to S3 as it exceeds the message_size_threshold limit