mcpField
S
sns

sns

github.com/itsmostafa云与 DevOps1.2k112 次安装2 天前更新
代码仓库

关于 sns

面向使用 Amazon SNS 进行发布/订阅消息传递的开发者的参考技能，涵盖主题、订阅、消息筛选、CLI 与 boto3 示例及投递故障排查。

主要功能

  1. 01Standard 与 FIFO 主题
  2. 02多种订阅协议
  3. 03基于属性的消息筛选
  4. 04AWS CLI 和 boto3 示例
  5. 05向多个 SQS 队列扇出
  6. 06投递故障排查

适用场景

  1. 01向主题订阅者发布通知
  2. 02通过订阅筛选器路由消息
  3. 03将消息扇出到 SQS 队列
  4. 04调查订阅端未收到消息的原因

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/itsmostafa/aws-agent-skills --skill sns

安装后在 Claude Code 中输入/sns确认已加载。

附带文件notification-patterns.md

详细介绍

AWS SNS

Amazon Simple Notification Service（SNS）是一项完全托管的发布/订阅消息服务，用于应用程序到应用程序（A2A）和应用程序到个人（A2P）的通信。

目录

核心概念

主题

用于发布消息的命名通道。发布者向主题发送消息，订阅者从主题接收消息。

主题类型
类型描述用例
Standard尽力保证顺序，至少传递一次大多数用例
FIFO严格保证顺序，恰好传递一次对顺序敏感的场景
订阅协议
协议描述
Lambda调用 Lambda 函数
SQS发送到 SQS 队列
HTTP/HTTPS向端点发送 POST 请求
Email发送电子邮件
SMS发送短信
Application移动推送通知
消息筛选

根据消息属性将消息路由到特定订阅者。

常见模式

创建主题并订阅

AWS CLI：

# Create standard topic
aws sns create-topic --name my-topic

# Create FIFO topic
aws sns create-topic \
  --name my-topic.fifo \
  --attributes FifoTopic=true

# Subscribe Lambda
aws sns subscribe \
  --topic-arn arn:aws:sns:us-east-1:123456789012:my-topic \
  --protocol lambda \
  --notification-endpoint arn:aws:lambda:us-east-1:123456789012:function:my-function

# Subscribe SQS
aws sns subscribe \
  --topic-arn arn:aws:sns:us-east-1:123456789012:my-topic \
  --protocol sqs \
  --notification-endpoint arn:aws:sqs:us-east-1:123456789012:my-queue

# Subscribe email
aws sns subscribe \
  --topic-arn arn:aws:sns:us-east-1:123456789012:my-topic \
  --protocol email \
  --notification-endpoint [email protected]

boto3：

import boto3

sns = boto3.client('sns')

# Create topic
response = sns.create_topic(Name='my-topic')
topic_arn = response['TopicArn']

# Subscribe Lambda
sns.subscribe(
    TopicArn=topic_arn,
    Protocol='lambda',
    Endpoint='arn:aws:lambda:us-east-1:123456789012:function:my-function'
)

# Subscribe SQS with filter
sns.subscribe(
    TopicArn=topic_arn,
    Protocol='sqs',
    Endpoint='arn:aws:sqs:us-east-1:123456789012:order-queue',
    Attributes={
        'FilterPolicy': '{"event_type": ["order_created", "order_updated"]}'
    }
)
发布消息
import boto3
import json

sns = boto3.client('sns')
topic_arn = 'arn:aws:sns:us-east-1:123456789012:my-topic'

# Simple publish
sns.publish(
    TopicArn=topic_arn,
    Message='Hello, World!',
    Subject='Notification'
)

# Publish with attributes (for filtering)
sns.publish(
    TopicArn=topic_arn,
    Message=json.dumps({'order_id': '12345', 'status': 'created'}),
    MessageAttributes={
        'event_type': {
            'DataType': 'String',
            'StringValue': 'order_created'
        },
        'priority': {
            'DataType': 'Number',
            'StringValue': '1'
        }
    }
)

# Publish to FIFO topic
sns.publish(
    TopicArn='arn:aws:sns:us-east-1:123456789012:my-topic.fifo',
    Message=json.dumps({'order_id': '12345'}),
    MessageGroupId='order-12345',
    MessageDeduplicationId='unique-id'
)
消息筛选
# Add filter policy to subscription
aws sns set-subscription-attributes \
  --subscription-arn arn:aws:sns:us-east-1:123456789012:my-topic:abc123 \
  --attribute-name FilterPolicy \
  --attribute-value '{
    "event_type": ["order_created"],
    "priority": [{"numeric": [">=", 1]}]
  }'

筛选策略示例：

// Exact match
{"event_type": ["order_created", "order_updated"]}

// Prefix match
{"customer_id": [{"prefix": "PREMIUM-"}]}

// Numeric comparison
{"price": [{"numeric": [">=", 100, "<=", 500]}]}

// Exists check
{"customer_id": [{"exists": true}]}

// Anything but
{"event_type": [{"anything-but": ["deleted"]}]}

// Combined
{
  "event_type": ["order_created"],
  "region": ["us-east", "us-west"],
  "priority": [{"numeric": [">=", 1]}]
}
扇出模式（SNS 到多个 SQS）
import boto3
import json

sns = boto3.client('sns')
sqs = boto3.client('sqs')

# Create topic
topic = sns.create_topic(Name='orders-topic')
topic_arn = topic['TopicArn']

# Create queues for different processors
queues = {
    'analytics': sqs.create_queue(QueueName='order-analytics')['QueueUrl'],
    'fulfillment': sqs.create_queue(QueueName='order-fulfillment')['QueueUrl'],
    'notification': sqs.create_queue(QueueName='order-notification')['QueueUrl']
}

# Subscribe each queue
for name, queue_url in queues.items():
    queue_arn = sqs.get_queue_attributes(
        QueueUrl=queue_url,
        AttributeNames=['QueueArn']
    )['Attributes']['QueueArn']

    sns.subscribe(
        TopicArn=topic_arn,
        Protocol='sqs',
        Endpoint=queue_arn
    )

# One publish reaches all queues
sns.publish(
    TopicArn=topic_arn,
    Message=json.dumps({'order_id': '12345', 'total': 99.99})
)
SNS 的 Lambda 权限
aws lambda add-permission \
  --function-name my-function \
  --statement-id sns-trigger \
  --action lambda:InvokeFunction \
  --principal sns.amazonaws.com \
  --source-arn arn:aws:sns:us-east-1:123456789012:my-topic

CLI 参考

主题管理
命令描述
aws sns create-topic创建主题
aws sns delete-topic删除主题
aws sns list-topics列出主题
aws sns get-topic-attributes获取主题设置
aws sns set-topic-attributes更新主题设置
订阅
命令描述
aws sns subscribe创建订阅
aws sns unsubscribe移除订阅
aws sns list-subscriptions列出所有订阅
aws sns list-subscriptions-by-topic列出主题订阅
aws sns confirm-subscription确认待处理的订阅
发布
命令描述
aws sns publish发布消息

最佳实践

可靠性
  • 使用 SQS 确保持久性 — SNS 基于推送，SQS 队列会对消息进行排队
  • 为 HTTP/HTTPS 端点实现重试
  • 为投递失败的消息配置 DLQ
  • 对有顺序要求的场景使用 FIFO 主题
安全性
  • 使用主题策略控制访问
  • 启用 SSE 加密
  • 使用 VPC 端点进行私有访问
# Enable SSE
aws sns set-topic-attributes \
  --topic-arn arn:aws:sns:us-east-1:123456789012:my-topic \
  --attribute-name KmsMasterKeyId \
  --attribute-value alias/my-key
成本优化
  • 使用消息筛选减少不必要的投递
  • 尽可能批量执行操作
  • 监控并清理未使用的主题/订阅
消息设计
  • 保持消息简短（上限为 256 KB）
  • 使用消息属性进行路由
  • 包含关联 ID以便追踪

故障排查

订阅未收到消息

检查：

  1. 订阅已确认（而非待处理状态）
  2. 筛选策略与消息属性匹配
  3. 目标权限（Lambda、SQS）
# Check subscription status
aws sns list-subscriptions-by-topic \
  --topic-arn arn:aws:sns:us-east-1:123456789012:my-topic

# Check subscription attributes
aws sns get-subscription-attributes \
  --subscription-arn arn:aws:sns:us-east-1:123456789012:my-topic:abc123
HTTP 端点无法正常工作

调试：

# Check delivery status logging
aws sns set-topic-attributes \
  --topic-arn arn:aws:sns:us-east-1:123456789012:my-topic \
  --attribute-name DeliveryPolicy \
  --attribute-value '{
    "http": {
      "defaultHealthyRetryPolicy": {
        "minDelayTarget": 20,
        "maxDelayTarget": 20,
        "numRetries": 3,
        "numMaxDelayRetries": 0,
        "numNoDelayRetries": 0,
        "numMinDelayRetries": 0,
        "backoffFunction": "linear"
      }
    }
  }'
消息与筛选条件不匹配

验证：

  • 已设置消息属性（而非消息正文中的属性）
  • 属性类型匹配（String 与 Number）
  • 筛选策略语法正确
# Correct: attributes must be message attributes
sns.publish(
    TopicArn=topic_arn,
    Message='body content',
    MessageAttributes={
        'event_type': {
            'DataType': 'String',
            'StringValue': 'order_created'  # This is filtered
        }
    }
)

# Wrong: this won't be filtered
sns.publish(
    TopicArn=topic_arn,
    Message=json.dumps({'event_type': 'order_created'})  # Not filtered
)
SQS 未从 SNS 接收消息

检查 SQS 队列策略：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {"Service": "sns.amazonaws.com"},
      "Action": "sqs:SendMessage",
      "Resource": "arn:aws:sqs:us-east-1:123456789012:my-queue",
      "Condition": {
        "ArnEquals": {
          "aws:SourceArn": "arn:aws:sns:us-east-1:123456789012:my-topic"
        }
      }
    }
  ]
}

参考资料

评论

支持 Markdown