1. 确保SNS主题的订阅是正确的且已验证。可以通过以下代码检查订阅是否存在并已验证：

import boto3

sns = boto3.client('sns') subscriptions = sns.list_subscriptions()['Subscriptions'] for subscription in subscriptions: if subscription['Protocol'] == 'sqs' and subscription['TopicArn'] == 'YOUR_TOPIC_ARN' and subscription['SubscriptionArn'] != 'PendingConfirmation': print('Subscription is verified and active')

2. 检查SQS队列的访问权限是否正确。确保已给SNS主题授予向SQS队列发送消息的权限。可以使用以下代码查看SNS主题的访问策略：

import boto3

sns = boto3.client('sns') policy = sns.get_topic_attributes(TopicArn='YOUR_TOPIC_ARN')['Attributes']['Policy'] print(policy)

如果策略未设置正确的访问权限，可以使用以下代码为其设置访问策略：

import boto3 from botocore.exceptions import ClientError

sns = boto3.client('sns')

policy = { "Version": "2008-10-17", "Id": "example", "Statement": [ { "Sid": "example-statement-ID", "Effect": "Allow", "Principal": { "AWS": "*" }, "Action": "sns:Publish", "Resource": "YOUR_TOPIC_ARN", "Condition": { "ArnEquals": { "aws:SourceArn": "YOUR_TOPIC_ARN" } } } ] }

try: sns.set_topic_attributes(TopicArn='YOUR_TOPIC_ARN', AttributeName='Policy', AttributeValue=json.dumps(policy)) except ClientError as e: print('Error: ', e)

3. 检查SQS队列的正确性。确保使用正确的SQS队列URL来订阅SNS主题。可以使用以下代码检查队列是否存在：

import boto3

sqs = boto3.client('sqs')

try: queue_url = sqs.get_queue_url(QueueName='YOUR_QUEUE_NAME')['QueueUrl'] print(queue_url) except: print('Queue does not exist')

如果队列不存在，可以使用以下代码创建一个新队列：

import boto3

sqs = boto3.client('sqs')

response = sqs.create_queue( QueueName='YOUR_QUEUE_NAME' )

queue_url = response['QueueUrl'] print(queue_url)

4. 检查SNS主题的发布程序