Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

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

亚马逊 SNS 电子邮件订阅中国地区终止支持

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

替代解决方案

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

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

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

代码示例：带有 Lambda 函数 SendCloud

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

先决条件

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 by using 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'}

迁移脚本 (Python)：将 Amazon SNS 订阅导出到 SendCloud

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

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!")

需要帮助或有疑问？

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

相关资源