Amazon SNS 电子邮件订阅已终止对中国地区的支持

替代解决方案代码示例：带有 Lambda 函数的代码 SendCloud 迁移脚本 (Python)：将亚马逊 SNS 订阅导出到 SendCloud 需要帮助或有疑问？相关资源

经过深思熟虑，我们决定自2027年6月30日起停止在由光环新网运营的 Amazon 中国（北京）地区和 Amazon 由西云数据运营的中国（宁夏）地区订阅Amazon SNS主题的电子邮件。在下述日期之后，Amazon SNS 将不再接受新的电子邮件订阅或向这些地区的电子邮件终端节点发送消息。

时间轴
日期	预计将发生的情况
2026 年 6 月 30 日	公开公告。所有客户都会收到通知。
2026年7月30日	Amazon 在这些地区没有电子邮件订阅的账户无法再向 Amazon SNS 主题添加电子邮件订阅。
2026 年 12 月 31 日	所有 Amazon 账户（包括已订阅电子邮件的账户）都无法再创建 Amazon SNS 主题的新电子邮件订阅。现有的电子邮件订阅继续有效。
2027年6月30日	所有电子邮件订阅均已删除。Amazon SNS 不再向电子邮件终端节点发送消息。客户会收到一份 Amazon Health 通知，其中包含其订阅的电子邮件地址的文件。

Amazon 不在中国（北京）地区或 Amazon Amazon 中国（宁夏）地区提供任何其他电子邮件投递服务。要继续通过 Amazon SNS 向电子邮件收件人发送消息，我们建议您订阅现有的 Amazon SNS 主题的 Amazon Lambda 功能，该主题可将消息转发给在中国运营的第三方电子邮件服务提供商。

您可以在 Amazon Marketplace 上找到第三方电子邮件服务提供商，例如SendCloud。

注意

Amazon 不认可第三方服务或为其提供支持。客户有责任根据其对交付能力、合规性、定价和区域可用性的要求对供应商进行评估。

要设置向第三方电子邮件提供商发送消息的 Lambda 函数，请执行以下步骤：

导出您订阅的电子邮件地址。在 2027 年 6 月 30 日之前，使用 Amazon SNS 控制台或 ListSubscriptionsByTopicAPI 导出订阅您的主题的电子邮件地址。2027 年 6 月 30 日之后，所有电子邮件订阅都将被删除，您可以从当天发送的 Healt Amazon h Notification 中检索已订阅的电子邮件地址。
选择第三方电子邮件提供商。选择符合您要求的提供商。请参阅上面的提供商列表。注册并完成提供商的入职流程。
将您的收件人列表导入提供商。大多数提供商都支持将电子邮件地址导入为联系人列表或地址列表。例如，使用 SendCloud，您可以使用地址列表 API 创建列表并以编程方式添加成员，也可以通过 CSV 导入。
创建一个 Lambda 函数。创建一个 Lambda 函数，用于接收 Amazon SNS 消息有效负载，并调用您的电子邮件提供商的 API 将消息发送到您的收件人列表。
将 Lambda 函数订阅到您的亚马逊 SNS 主题。使用亚马逊 SNS 控制台或 API。Subscribe
测试端到端流程。向您的主题发布测试消息，并通过提供商验证电子邮件的传送情况。
移除电子邮件订阅。确认迁移后，您可以使用 Amazon SNS 控制台或 API 删除电子邮件订阅。Unsubscribe所有剩余的电子邮件订阅将于 2027 年 6 月 30 日自动删除。

以下示例显示了一个 Lambda 函数，该函数接收来自 Amazon SNS 主题的消息并使用 SendCloud 地址列表将其作为电子邮件发送。

先决条件

具有经过验证的发件人域名的 SendCloud 账户
在中创建的 SendCloud 包含您的收件人的地址列表（从您的 Amazon SNS 电子邮件订阅中导入）
配置了以下 Lambda 环境变量：
- SENDCLOUD_API_USER: 你的 SendCloud API 用户
- SENDCLOUD_API_KEY: 你的 SendCloud API 密钥
- SENDER_EMAIL: 您经过验证的发件人电子邮件地址（例如 notifications@yourdomain.com）
- ADDRESS_LIST: 你的 SendCloud 地址列表别名（例如，mylist@maillist.sendcloud.org）

注意

这个简化的示例从 Lambda 环境变量中读取证书。在生产环境中，可以考虑将凭据存储在 S Amazon ecrets Manager 中。

Lambda 函数 (Python)


import json
import urllib.request
import urllib.parse
import os

# SendCloud credentials. This simplified example reads them from Lambda
# environment variables; in production, store them in Amazon Secrets Manager.
API_USER = os.environ['SENDCLOUD_API_USER']
API_KEY = os.environ['SENDCLOUD_API_KEY']
SENDER_EMAIL = os.environ['SENDER_EMAIL']  # e.g., notifications@yourdomain.com
ADDRESS_LIST = os.environ['ADDRESS_LIST']   # e.g., mylist@maillist.sendcloud.org

# SendCloud API endpoint
SENDCLOUD_API_URL = 'https://api.sendcloud.net/apiv2/mail/send'


def lambda_handler(event, context):
    """
    Receives an SNS message and sends it as an email via SendCloud.
    """
    for record in event['Records']:
        sns_message = record['Sns']
        subject = sns_message.get('Subject', 'Notification')
        message_body = sns_message.get('Message', '')
        timestamp = sns_message.get('Timestamp', '')
        topic_arn = sns_message.get('TopicArn', '')

        # Build email content
        html_body = f"""
        <html>
        <body>
            <p>{message_body}</p>
            <hr>
            <p style="font-size: 12px; color: #666;">
                Source: {topic_arn}<br>
                Time: {timestamp}
            </p>
        </body>
        </html>
        """

        # Send via SendCloud API
        params = {
            'apiUser': API_USER,
            'apiKey': API_KEY,
            'from': SENDER_EMAIL,
            'to': ADDRESS_LIST,
            'useAddressList': 'true',
            'subject': subject,
            'html': html_body,
        }

        data = urllib.parse.urlencode(params).encode('utf-8')
        req = urllib.request.Request(SENDCLOUD_API_URL, data=data, method='POST')

        try:
            with urllib.request.urlopen(req) as response:
                result = json.loads(response.read().decode('utf-8'))
                if result.get('result'):
                    print(f"Email sent successfully: {result.get('message')}")
                else:
                    print(f"SendCloud error: {result}")
                    raise Exception(f"SendCloud API error: {result.get('message')}")
        except Exception as e:
            print(f"Failed to send email: {str(e)}")
            raise  # Let Lambda retry or send to DLQ

    return {'statusCode': 200, 'body': 'Messages processed'}

以下脚本导出来自 Amazon SNS 主题的电子邮件订阅并将其导入到 SendCloud 地址列表中。在运行之前，请配置您的 Amazon 区域、Amazon SNS 主题 ARN、 SendCloud证书和地址列表。

注意

这个简化的示例在脚本中嵌入了凭证。在生产环境中，可以考虑将凭据存储在 S Amazon ecrets Manager 中。


import boto3
import urllib.request
import urllib.parse
import json

# Configuration — update these values before running
REGION = 'cn-north-1'  # or 'cn-northwest-1' for Ningxia
TOPIC_ARN = 'arn:aws-cn:sns:cn-north-1:<YOUR_ACCOUNT_ID>:<YOUR_TOPIC_NAME>'

# SendCloud credentials. This is a simplified example; in production, store
# them in Amazon Secrets Manager rather than inline in the script.
API_USER = '<YOUR_SENDCLOUD_API_USER>'
API_KEY = '<YOUR_SENDCLOUD_API_KEY>'

# your SendCloud address list alias
ADDRESS_LIST = '<YOUR_ADDRESS_LIST>@maillist.sendcloud.org'

# Amazon SDK
sns_client = boto3.client('sns', region_name=REGION)

SENDCLOUD_ADD_MEMBER_URL = 'https://api.sendcloud.net/apiv2/addressmember/add'


def get_email_subscriptions(topic_arn):
    """Retrieve all email subscriptions from an SNS topic."""
    emails = []
    paginator = sns_client.get_paginator('list_subscriptions_by_topic')
    for page in paginator.paginate(TopicArn=topic_arn):
        for sub in page['Subscriptions']:
            if sub['Protocol'] in ('email', 'email-json'):
                emails.append(sub['Endpoint'])
    return emails


def add_to_sendcloud(emails, batch_size=1000):
    """Add email addresses to a SendCloud address list in batches."""
    for i in range(0, len(emails), batch_size):
        batch = emails[i:i + batch_size]
        members = ';'.join(batch)

        params = {
            'apiUser': API_USER,
            'apiKey': API_KEY,
            'address': ADDRESS_LIST,
            'members': members,
        }

        data = urllib.parse.urlencode(params).encode('utf-8')
        req = urllib.request.Request(SENDCLOUD_ADD_MEMBER_URL, data=data, method='POST')

        with urllib.request.urlopen(req) as response:
            result = json.loads(response.read().decode('utf-8'))
            if result.get('result'):
                print(f"Added {len(batch)} members (batch {i // batch_size + 1})")
            else:
                print(f"Error adding batch: {result}")


if __name__ == '__main__':
    print(f"Exporting email subscriptions from {TOPIC_ARN}...")
    emails = get_email_subscriptions(TOPIC_ARN)
    print(f"Found {len(emails)} email subscriptions")

    print(f"Importing into SendCloud address list: {ADDRESS_LIST}...")
    add_to_sendcloud(emails)
    print("Migration complete!")

如果您需要帮助或有反馈，请联系 Supp Amazon ort。

Javascript 在您的浏览器中被禁用或不可用。

要使用 Amazon Web Services 文档，必须启用 Javascript。请参阅浏览器的帮助页面以了解相关说明。

电子邮件订阅设置和管理

代码示例