AWSSQS队列未从SNS主题订阅接收消息。
- 确保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')
- 检查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)
- 检查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)
- 检查SNS主题的发布程序