在微服务架构里服务之间的通信方式直接决定整个系统的稳定性边界。很多团队一开始习惯用同步 HTTP 调用完成所有业务协作比如订单服务创建订单后直接调用库存服务扣减库存、调用通知服务发送邮件、调用积分服务增加积分。这种模式在流量小、链路短的时候没有问题一旦服务增多、流量上涨同步链路会把下游服务的抖动逐级放大。AWS SNS 和 SQS 正是为了解决这类问题而存在的消息组件SNS 负责把事件广播给多个订阅方SQS 负责把消息积压、削峰并等待消费者按需处理。本系列第一篇会从概念讲起用一个最小可运行案例把“SNS 发布事件 - SQS 接收事件 - 消费者处理事件”的完整链路跑通随后解释可靠性设计、死信队列和常见排查思路。读完这篇后你能在自己的微服务项目里判断哪类场景该用 SNS、哪类场景该用 SQS也能独立搭出一条事件驱动链路。1. 为什么微服务架构需要消息中间件而不是只靠 HTTP 同步调用1.1 同步调用在微服务链路中的四个问题微服务之间的同步调用不是不能使用而是需要认识到它的代价。假设下单接口内部要依次调用 4 个下游服务整个接口的响应时间就是 4 个下游响应时间之和再加上网关、序列化和网络传输时间。任何一个下游服务只要出现 300 毫秒延迟下单接口就会从原本的 50 毫秒变成 350 毫秒。这个延迟叠加问题在调用链变深后会非常明显。更麻烦的是可用性问题。4 个下游服务只要有一个不可用下单主流程可能直接失败或超时。即使服务之间配置了超时时间和重试机制同步重试也会把流量继续压向下游。当一个服务宕机依赖它的所有上游都会因为重试而增加积压请求最终形成雪崩效应。还有一类情况是业务上根本不需要立即返回结果。订单创建成功后系统要发送邮件、生成推荐数据、更新报表、推送消息给仓配系统。用户并不需要等待这些动作全部完成才能看到“下单成功”的响应。如果把这些动作放进同步链等于把非核心路径的延迟强加给了核心路径。第四个问题是流量峰值。电商大促、秒杀、活动抢购都会出现短时间的高并发。下游服务如果按照峰值流量扩容成本很高如果不扩容又会在高峰期被打垮。消息队列可以在上游和下游之间增加一个缓冲层让瞬时流量先进入队列再由消费者按自己的处理能力把消息消费完。1.2 SNS 和 SQS 的分工一个负责广播一个负责缓冲SNS 的全称是 Simple Notification Service定位是发布/订阅模型。生产者把消息发布到 TopicTopic 会把消息推送给所有订阅者。订阅者可以是 SQS 队列、Lambda 函数、HTTP 端点、邮件地址、移动终端等。SNS 不负责缓存消息给消费者慢慢取它更多是事件分发器。SQS 的全称是 Simple Queue Service定位是消息队列。生产者把消息发送到 Queue消费者主动轮询拉取消息处理成功后删除消息。SQS 会把消息保留一段时间默认保留 4 天最长可以配置成 14 天。消费者不在线时消息会一直留在队列里由后来启动的消费者继续处理。在微服务架构中SNS 和 SQS 的组合非常常见。一个订单创建事件发布到 SNS Topic 后SNS 可以把消息推送给“库存服务的 SQS 队列”“通知服务的 SQS 队列”“数据仓库的 SQS 队列”。每个下游服务只消费自己的队列互不干扰。这样不仅把事件分发和异步处理分开还让下游服务在消费时拥有独立的重试和积压能力。1.3 一条典型的事件驱动链路一个典型的订单场景可以这样设计订单服务在数据库事务提交成功后发布ORDER_CREATED事件到 SNS Topic。SNS 根据订阅关系把事件推送给 N 个 SQS 队列。库存服务从自己的队列中拉取事件扣减库存通知服务从自己的队列拉取事件发送消息分析服务从自己的队列拉取事件写入数据仓库。发送到 SNS 的消息并不会占用订单事务时间。订单服务只需要把消息发布成功就算完成了事件通知。下游服务如果暂时不可用消息会留在 SQS 队列中下游恢复后消费者会继续从队列中拉取并处理。这样整个系统从同步调用变成了异步事件驱动核心链路稳定性不再受非核心服务影响。2. 先把 SNS 和 SQS 的机制弄明白再动手配置2.1 SNS Topic发布/订阅模型背后是推送关系SNS 的核心概念是 Topic。一个 Topic 相当于一个广播频道。生产者把消息 publish 到 TopicTopic 会把消息推送给所有订阅者。订阅者之间是独立的一个订阅者处理失败不会影响另一个订阅者。SNS 的消息结构里包含几个关键字段Type表示消息类型一般是NotificationMessageId是消息唯一标识TopicArn是主题的 ARNMessage是业务负载Timestamp是发布时间如果需要还可以包含MessageAttributes用于描述消息格式、消息标签等。订阅方收到的就是一个 JSON 格式的消息壳。这里容易误解的点是SNS 不会等待订阅者消费成功后再返回。SNS 的发布 API 只需要确认消息已经投递给订阅通道比如已经交给 SQS 队列就算成功。至于 SQS 里的消息何时被消费者处理SNS 不关心。2.2 SQS Queue为什么消费者要主动拉取而不是被动接收SQS 和常见的消息队列中间件不同它默认采用拉模型。消费者调用ReceiveMessage从队列里取一批消息取到之后这些消息会进入不可见状态其他消费者在这个不可见期间无法取到。消息处理成功后消费者调用DeleteMessage删除该消息如果处理失败消费者可以不删除消息会在可见性超时结束后重新出现在队列里。这个机制保证了消息不会被多个消费者同时拉取也允许消费者根据自身负载能力控制消费速率。消费者如果处理速度不够队列中的消息会积压但不会丢失。通过配置长轮询消费者可以设置WaitTimeSeconds让 SQS 在没有消息时等待最长 20 秒再返回空响应从而减少空轮询带来的 API 请求数量。SQS 默认提供标准队列它的语义是至少一次投递也就是说消息可能被重复投递。另一个类型是 FIFO 队列它提供严格顺序和恰好一次语义但每秒吞吐量有限制。对于大多数微服务事件转发场景标准队列就够用但业务处理逻辑必须设计成支持重复消息。2.3 SNS 和 SQS 的选型对照在实际项目中SNS 和 SQS 不是二选一的关系。它们解决的问题不同经常叠加使用。下面这张表格可以帮助快速判断场景。场景推荐方案原因一个订单事件要发给库存、通知、报表等多个下游SNS Topic 多个 SQS QueueSNS 负责分发SQS 为每个下游提供独立缓冲只有一个消费者消息要异步慢慢处理SQS 单队列使用队列缓存流量峰值消费者按需拉取需要推送给移动设备或邮件SNS 直接订阅移动端或邮箱SNS 原生支持多种推送协议需要严格顺序处理某个业务实体的消息SQS FIFO 队列FIFO 保证同 GroupId 内的消息顺序需要事件广播给 Lambda 做实时处理SNS 订阅 LambdaLambda 被触发后可以直接处理事件某条消息需要延迟一段时间后再处理SQS DelaySeconds 参数生产者发送时或队列配置中设置延迟时间这里要特别说明一个 SNS Topic 可以直接订阅 SQS 队列也可以订阅多个 SQS 队列。这是微服务事件驱动中最常见的组合方式。不要在架构设计里把 SNS 和 SQS 当成互相替代品它们组合起来才能兼顾广播和缓冲。2.4 为什么 SNS 直接订阅 SQS 比“业务服务自己轮询 SNS”合适有些第一次接触事件驱动的同学会问能不能只用一个 SNS Topic让每个服务自己轮询 SNS这里有个概念误区。SNS 不是“消息仓库”它不会保存消息供消费者随时来取消息发布后只负责推送给订阅端。如果某个订阅端不在线SNS 对 HTTP 这类协议会有重试策略但不会长期存储消息。SQS 才是承担消息存储和积压的组件。所以正确的事件驱动架构是SNS 负责“把事件告诉所有感兴趣的订单方”SQS 负责“把事件安全地保存到订单方自己的队列里”。每个订单方从自己的 SQS 队列拉取事件处理完成后删除消息。这套组合让消息既有广播能力又有持久化缓冲能力。3. 搭建实验环境准备账号、CLI、SNS 主题和 SQS 队列3.1 环境准备与前置知识要跟着完整跑通本文案例你需要准备以下内容一个 AWS 账号一双有权限创建 SNS 和 SQS 资源的 IAM 用户或角色本机安装并配置好 AWS CLI本机安装 Python 3 和 boto3一个用于实验的 Region比如us-east-1。先确认本机 AWS CLI 已经配置成功。aws --version aws sts get-caller-identity正常输出会包含你的账号 ID、ARN 和用户名称。如果返回Unable to locate credentials需要先执行aws configure配置Access Key ID、Secret Access Key、默认 Region 和输出格式。接下来安装 Python 依赖后续消费者脚本会用到 boto3。pip install boto3这里要注意学习环境可以使用权限较大的凭证但生产环境一定要按最小权限原则分配 IAM 策略。文中后续会给生产环境的最小权限示例。3.2 通过 CLI 创建 SQS 队列和 SNS 主题创建实验资源时先定义几个变量。export AWS_REGIONus-east-1 export TOPIC_NAMEorder-events-topic export QUEUE_NAMEorder-events-queue创建 SNS 主题。aws sns create-topic --name $TOPIC_NAME --region $AWS_REGION创建 SQS 队列配置可见性超时和消息保留时间。aws sqs create-queue \ --queue-name $QUEUE_NAME \ --region $AWS_REGION \ --attributes { VisibilityTimeout: 30, MessageRetentionPeriod: 86400, ReceiveMessageWaitTimeSeconds: 20 }参数说明VisibilityTimeout是消息被消费者拉取后的不可见时长这里设置为 30 秒MessageRetentionPeriod是消息最大保留时间这里设置为 86400 秒也就是 1 天ReceiveMessageWaitTimeSeconds是长轮询的等待时间设置成 20 秒可以减少空轮询。查询创建出来的主题 ARN 和队列 ARN。后续配置订阅时需要用到。aws sns list-topics --region $AWS_REGION aws sqs list-queues --region $AWS_REGION aws sqs get-queue-attributes \ --queue-url https://sqs.$AWS_REGION.amazonaws.com/123456789012/$QUEUE_NAME \ --attribute-names QueueArn \ --region $AWS_REGION注意将队列 URL 中的123456789012替换成自己的账号 ID。最简单的做法是先执行aws sqs list-queues拿到完整队列 URL再执行get-queue-attributes。3.3 创建 SNS 到 SQS 的订阅并配置访问策略光有主题和队列还不够需要在两者之间建立订阅关系。SNS 要把消息推送到 SQS必须把 SQS 队列的 ARN 作为订阅端点。aws sns subscribe \ --topic-arn arn:aws:sns:$AWS_REGION:123456789012:$TOPIC_NAME \ --protocol sqs \ --notification-endpoint arn:aws:sqs:$AWS_REGION:123456789012:$QUEUE_NAME \ --region $AWS_REGION订阅命令执行后可以通过下面的命令查看订阅状态。aws sns list-subscriptions-by-topic \ --topic-arn arn:aws:sns:$AWS_REGION:123456789012:$TOPIC_NAME \ --region $AWS_REGION订阅创建初期SubscriptionArn可能是PendingConfirmation当 SNS 成功向 SQS 推送消息后订阅状态会变成已确认。这里的“推送”并不是说 SNS 正在推业务消息而是 SNS 需要验证自己是否具备向 SQS 发送消息的权限。如果队列策略不允许 SNS 发送订阅会一直处于等待确认状态消息也无法进入 SQS。因此需要给 SQS 队列配置一条基于sns.amazonaws.com的发送策略。下面是通过 Python 脚本设置队列策略的方式便于直接写入 JSON。import boto3 import json client boto3.client(sqs, region_nameus-east-1) queue_url https://sqs.us-east-1.amazonaws.com/123456789012/order-events-queue topic_arn arn:aws:sns:us-east-1:123456789012:order-events-topic queue_arn arn:aws:sqs:us-east-1:123456789012:order-events-queue policy { Version: 2012-10-17, Statement: [ { Effect: Allow, Principal: {Service: sns.amazonaws.com}, Action: sqs:SendMessage, Resource: queue_arn, Condition: { ArnEquals: { aws:SourceArn: topic_arn } } } ] } client.set_queue_attributes( QueueUrlqueue_url, Attributes{Policy: json.dumps(policy)} ) print(Queue policy set.)这里使用ArnEquals条件把来源限制到指定 SNS Topic而不是允许所有 SNS 主题都往该队列发消息。这种做法在多个业务共用账号的微服务环境中非常重要否则一个其他业务的 SNS 主题也能往这个队列写消息会导致消息污染。检查订阅确认状态。aws sns get-subscription-attributes \ --subscription-arn arn:aws:sns:us-east-1:123456789012:order-events-topic:你的订阅ID \ --region $AWS_REGION如果SubscriptionArn是完整的 ARN并且PendingConfirmation不再出现说明订阅已经生效。3.4 验证订阅状态的小清单完成这一步后用下面的清单确认所有基础资源都正常。检查项预期结果检查命令SNS Topic 存在返回 TopicArnaws sns list-topicsSQS 队列存在返回 QueueUrlaws sqs list-queues订阅状态已确认SubscriptionArn 不是 PendingConfirmationaws sns list-subscriptions-by-topic队列策略已配置策略中包含 SNS 发送权限aws sqs get-queue-attributes --attribute-names Policy这一节完成后SNS 和 SQS 之间的通道已经打通。下一步就是发布消息并且消费它。4. 把消息流跑通一个订单事件从发布到消费4.1 设计事件结构进入代码之前先约定事件结构。事件是服务之间传递的语义单位推荐包含事件类型、事件ID、业务实体的关键字段和发生时间。{ eventId: evt_10001, eventType: ORDER_CREATED, orderId: order_20260001, amount: 299.00, createdAt: 2026-01-01T10:00:00Z }eventId是事件的唯一标识可以用来做幂等去重eventType是事件的语义类型orderId是业务实体主键amount是订单金额createdAt是事件发生时间。事件结构应尽量只包含消费者做决策所必需的数据不要塞入数据库表结构等内部细节。4.2 发布消息到 SNS Topic发布消息可以使用 CLI也可以使用 Python。先看 CLI 方式。aws sns publish \ --topic-arn arn:aws:sns:us-east-1:123456789012:order-events-topic \ --message {eventId:evt_10001,eventType:ORDER_CREATED,orderId:order_20260001,amount:299.00,createdAt:2026-01-01T10:00:00Z} \ --region $AWS_REGION返回结果中会包含MessageId说明 SNS 已经接收消息。用 Python 发布时代码更接近实际生产项目的写法。import boto3 import json client boto3.client(sns, region_nameus-east-1) response client.publish( TopicArnarn:aws:sns:us-east-1:123456789012:order-events-topic, Messagejson.dumps({ eventId: evt_10001, eventType: ORDER_CREATED, orderId: order_20260001, amount: 299.00, createdAt: 2026-01-01T10:00:00Z }) ) print(response[MessageId])这里 SNS 的Message参数虽然是字符串但业务上通常放入一段 JSON。消费者收到后需要先解析外层 SNS 消息壳再解析内层业务 JSON。4.3 用 Python 消费者从 SQS 拉取消息下面是一个最小可运行的消费者脚本它不断从 SQS 队列中拉取消息解析业务事件然后删除消息。import boto3 import json client boto3.client(sqs, region_nameus-east-1) queue_url https://sqs.us-east-1.amazonaws.com/123456789012/order-events-queue while True: response client.receive_message( QueueUrlqueue_url, MaxNumberOfMessages10, WaitTimeSeconds20, VisibilityTimeout30 ) for message in response.get(Messages, []): # 第一层SNS 推送通知的消息壳 envelope json.loads(message[Body]) # 第二层业务事件 event json.loads(envelope[Message]) print(收到事件类型:, event[eventType]) print(订单ID:, event[orderId]) print(金额:, event[amount]) # 业务处理成功后必须删除消息 client.delete_message( QueueUrlqueue_url, ReceiptHandlemessage[ReceiptHandle] )这段代码有几个关键点。第一次解析message[Body]得到的是 SNS 信封里面包含Type、MessageId、TopicArn、Message等字段。第二次解析envelope[Message]得到的是业务事件 JSON。ReceiptHandle是删除消息的唯一凭证。每次收到消息时都要使用当次返回的ReceiptHandle不能提前存下来长期复用。如果消费者在处理完业务后不调用delete_message这条消息会在可见性超时后再次出现在队列中导致重复消费。这里使用WaitTimeSeconds20的长轮询可以避免消费者在没有消息时频繁向 SQS 发起空请求。短轮询每个请求都会立即返回即使没有消息也会产生 API 费用长轮询能显著减少请求次数。4.4 验证消息链路是否已打通消费者脚本运行后再回到另一个终端发布一条消息。正常情况下消费者终端会输出收到事件类型: ORDER_CREATED 订单ID: order_20260001 金额: 299.0你还可以去 SQS 控制台查看队列的NumberOfMessagesSent和NumberOfMessagesReceived指标确认消息已经被 SQS 接收并且消费者已经拉取成功。如果消费者脚本运行后没有输出先检查订阅状态和队列策略再检查消息是否进入了死信队列或是否被其他消费者取走。在这个最小链路里订单服务就相当于发送方脚本库存服务或通知服务就相当于消费者脚本。两者之间不需要知道彼此的地址和接口只依赖 SNS Topic 和 SQS Queue 的 ARN。这就是事件驱动架构的解耦价值。5. 设计可靠的异步消息链路重试、死信与幂等5.1 SQS 不丢消息但可能重复投递消息SQS 标准队列的语义是至少一次投递。也就是说正常情况下每条消息至少会被消费一次但极端情况下可能重复。这与传统数据库事务的单次提交逻辑不一样不要拿“消息不会重复”的假设去设计业务。重复可能出现在两个地方一是 SQS 内部在极端情况下对同一条消息进行了多次可见性恢复二是消费者处理消息时超过了可见性超时时间消息在业务还没有处理完成时又变回可见被另一个消费者拉取。因此消费者代码不能只做“收到消息就处理、处理完就删除”这一套动作还需要在业务层面做到幂等。所谓幂等就是同一条业务事件即使被执行两次最终结果也和执行一次一样。5.2 消费失败时不删除消息让重试自然发生消费者处理业务时可能出现数据库连接失败、下游接口超时、数据格式不正确等问题。如果处理失败就调用delete_message消息会丢失如果处理不成功也不删除消息会在可见性超时结束后重新回到队列等待下一次消费。推荐的流程是先尝试执行业务逻辑业务成功后删除消息业务失败则不要删除消息让 SQS 在可见性超时后重新投递。这里要结合日志记录失败原因避免消息反复重试但没有任何排查线索。try: event json.loads(envelope[Message]) process_order(event) client.delete_message( QueueUrlqueue_url, ReceiptHandlemessage[ReceiptHandle] ) except Exception as exc: print(消息处理失败:, exc) # 不删除消息等待 SQS 重新投递不要在高频路径上直接打印完整消息内容要打印消息 ID、事件类型、失败原因方便后续定位。5.3 配置死信队列拦截多次重试的坏消息如果某条消息因为数据格式问题一直处理失败它会永远在队列里反复出现占用消费线程也影响队列健康。SQS 提供死信队列机制可以把多次重试仍然失败的消息转移到一个独立队列业务工程师可以从死信队列中单独排查。先创建一个死信队列。aws sqs create-queue \ --queue-name order-events-dlq \ --region $AWS_REGION获取死信队列的 ARN。aws sqs get-queue-attributes \ --queue-url https://sqs.us-east-1.amazonaws.com/123456789012/order-events-dlq \ --attribute-names QueueArn \ --region $AWS_REGION然后给主队列配置 RedrivePolicy。import boto3 import json client boto3.client(sqs, region_nameus-east-1) queue_url https://sqs.us-east-1.amazonaws.com/123456789012/order-events-queue dlq_arn arn:aws:sqs:us-east-1:123456789012:order-events-dlq client.set_queue_attributes( QueueUrlqueue_url, Attributes{ RedrivePolicy: json.dumps({ deadLetterTargetArn: dlq_arn, maxReceiveCount: 3 }) } )maxReceiveCount表示消息最多被消费者拉取 3 次。如果 3 次都失败消息会被转移到死信队列。这样主队列不会一直被坏消息阻塞运维人员可以从死信队列中恢复或修正这些消息。需要注意的是死信队列本身不能再配置指向原队列的 RedrivePolicy否则会形成循环投递。检测死信队列的消费者时通常需要人工介入修复数据后把消息重新投递回主队列或直接把消息归档。5.4 幂等处理在微服务里的落地方式幂等处理不是靠某个框架自动完成的需要业务层自己设计。常见做法是为每条业务事件分配一个唯一事件 ID在消费端记录已经处理过的事件 ID重复消息直接跳过。例如在订单事件里用eventId作为去重键。消费者在处理前先查询去重表或 Redis如果eventId已存在就直接删除消息不再执行业务逻辑。下面是一个伪代码示例。def process_event(event): event_id event[eventId] if redis_client.setnx(fdedup:{event_id}, 1, ex86400): process_order(event) return True return FalseSETNX表示只有键不存在时才写入成功。如果写入成功说明这条事件第一次出现可以执行业务逻辑如果写入失败说明重复消息直接返回。在数据库操作场景里更稳妥的方式是利用数据库唯一约束。比如把订单事件的处理结果表设计成eventId为主键重复插入会触发唯一冲突业务代码捕获冲突后按成功处理。这样即使消费者进程在去重步骤之后、业务提交之前崩溃也能在重启后通过唯一约束拦截重复数据。6. 消息链路常见问题排查与解决6.1 订阅状态是 PendingConfirmation或者一直收不到消息现象执行list-subscriptions-by-topic后SubscriptionArn显示为PendingConfirmation。向 SNS 发布消息后SQS 队列里没有消息。排查顺序检查订阅端点是否写成了 SQS 队列 URL而不是队列 ARN。SNS 订阅 SQS 的notification-endpoint必须是队列 ARN。检查队列策略是否允许 SNS 发送消息。缺少策略时SNS 无法完成订阅确认。检查 Topic 和 Queue 是否在同一个账号、同一个 Region。跨账号也可以配置但需要更复杂的资源策略。检查订阅的RawMessageDelivery属性。如果设置成 trueSQS 收到的消息里不会包含 SNS 信封消费端解析逻辑会不一样。解决方式补上队列策略重新确认订阅状态。对 SQS 订阅一般不需要手动确认订阅只要权限正确订阅 ARN 会自动从 PendingConfirmation 变为完整 ARN。6.2 消费者进程反复收到同一条消息现象消费者明明处理并删除了消息但过一会儿又收到同一条消息或者消息处理时间较长还没有删就已经被重新投递。常见原因有两个。第一个是消息处理时间超过了可见性超时。假设消费者处理一条消息需要 40 秒但队列的VisibilityTimeout只有 30 秒消息在 30 秒后重新可见其他消费者就能再次拉取到它。第二个原因是消费者处理失败后没有删除消息消息在可见性超时后重新回到队列。这种情况下需要看日志确认是业务失败还是删除失败。解决方式把队列的VisibilityTimeout设置成大于消费者单条消息处理时间对于耗时非常长的任务可以在处理过程中调用ChangeMessageVisibility延长不可见时间。6.3 消费者解析消息时报错或字段不存在现象消费端使用event[orderId]时报KeyError或TypeError。原因通常是解析层级不对。SNS 订阅 SQS 后队列中的消息是一个复合结构。常见错误是只解析了一层直接用json.loads(message[Body])[orderId]但订单 ID 在外层信封的Message字段里。解决方式先解析外层得到 SNS 信封再解析envelope[Message]得到业务消息。如果订阅开启了RawMessageDelivery则队列里直接就是 SNS 发布时的原始消息内容不需要再解析信封。两种模式必须保持一致否则代码会解析失败。6.4 从指标到日志的排查顺序消息链路涉及发布端、SNS、SQS、消费端四个环节排查时要按“消息是否发布 - 是否进入队列 - 是否被接收 - 是否被删除”的顺序推进。现象检查位置可能原因SNS 发布失败SNS 发布调用日志、IAM 权限缺少 sns:Publish 权限、TopicArn 拼写错误队列消息数为 0SQS 控制台指标订阅未确认、队列策略错误、发布端没有执行成功队列有消息但消费者不消费消费者日志、可见性超时消费者没有运行、IAM 缺少 ReceiveMessage 权限消息反复出现消费者日志、RedrivePolicy处理失败未删除、可见性超时过短、缺少幂等处理消息进入死信队列死信队列指标maxReceiveCount 设置过低、业务代码存在无法修复的数据问题AWS 控制台的 CloudWatch 指标可以直接看到NumberOfMessagesSent、NumberOfMessagesReceived、NumberOfMessagesDeleted、ApproximateAgeOfOldestMessage。这些指标可以帮助快速定位消息卡在哪个环节。7. 从实验环境走向生产环境还需要补齐哪些能力7.1 学习环境与生产环境的配置差异本文的演示只适合跑通链路直接照搬到生产环境会有明显风险。下面这张表整理了主要差异。事项学习环境生产环境凭证权限可使用管理员权限使用最小权限 IAM 策略区分发布者和消费者角色队列策略允许单 Topic 写入即可限定账号、Topic、Region配置更严格的 SourceArn 条件加密可不配置启用 SQS 加密使用 AWS KMS 托管密钥监控手动查控制台配置 CloudWatch 告警关注队列积压和消费失败错误处理print 日志结构化日志、错误追踪、死信队列人工处理流程消息结构手动维护 JSON尽量使用显式事件契约发布前校验字段幂等可能不做必须有事件 ID 和去重逻辑否则重复消息会造成数据错误生产环境里消费端至少还需要考虑配置外置化、日志采集、监控告警和异常逃逸处理。消费者进程要能被编排工具管理异常退出后自动重启不能再靠终端前台运行脚本。7.2 最小权限 IAM 策略示例发布者角色只需要调用 SNS Publish 的权限并且限制到指定 Topic。{ Version: 2012-10-17, Statement: [ { Effect: Allow, Action: sns:Publish, Resource: arn:aws:sns:us-east-1:123456789012:order-events-topic } ] }消费者角色只需要对指定队列执行接收、删除和修改可见性超时的权限。{ Version: 2012-10-17, Statement: [ { Effect: Allow, Action: [ sqs:ReceiveMessage, sqs:DeleteMessage, sqs:ChangeMessageVisibility ], Resource: arn:aws:sqs:us-east-1:123456789012:order-events-queue } ] }资源策略和身份策略要配合使用。队列策略控制“谁可以往队列发送消息”IAM 策略控制“某个身份可以执行哪些操作”。不要为了省事给开发人员配置对所有 SNS 和 SQS 资源的*权限。7.3 成本控制与监控要点SNS 和 SQS 的计费主要看请求次数和传输流量消息大小对成本也有影响。SQS 单条消息最大为 256 KBAPI 计费以 64 KB 为最小单位计算。消费者如果使用短轮询并且频繁调用ReceiveMessage空请求也会产生费用因此生产代码应优先使用长轮询。监控上重点关注三个指标ApproximateNumberOfMessagesVisible表示队列中可见消息数ApproximateAgeOfOldestMessage表示最老消息的滞留时间NumberOfMessagesReceived和NumberOfMessagesDeleted用于对照消费者处理情况。最老的滞留给消息年龄不断增长时说明消费者处理能力不足或消息一直失败。建议对主队列的死信队列配置告警一旦有消息进入死信队列就触发人工排查流程。这个告警是事件驱动系统里最重要的生产告警之一。7.4 本系列的下一步扩展方向这一篇解决了 SNS 和 SQS 的基础链路Topic、Queue、订阅、发布、消费、死信和幂等。配置好这套基础后你可以在自己的微服务项目里做两类扩展。第一类是把 SNS 消息本身加过滤策略让不同下游只接收自己关心的事件类型减少无效消息。第二类是用 Lambda 作为 SQS 事件的消费者利用 SQS 触发 Lambda 的能力省掉显式运行消费者进程的运维成本。如果业务对消息顺序有严格要求下一步应该研究 SQS FIFO 队列。FIFO 队列支持按MessageGroupId分组同一组内的消息严格有序且不会重复但吞吐量有上限适用范围和标准队列不同。这部分内容适合作为本系列的第二篇主题。在把更多服务接入事件驱动之前最值得投入的仍然是两件事把消息处理设计成幂等把死信消息的排查流程跑通。只有这两个基础点稳了依赖 SNS 和 SQS 的大规模异步链路才有资格进入生产环境。 微服务通信利器：AWS SNS与SQS消息队列入门与实践
在微服务架构里服务之间的通信方式直接决定整个系统的稳定性边界。很多团队一开始习惯用同步 HTTP 调用完成所有业务协作比如订单服务创建订单后直接调用库存服务扣减库存、调用通知服务发送邮件、调用积分服务增加积分。这种模式在流量小、链路短的时候没有问题一旦服务增多、流量上涨同步链路会把下游服务的抖动逐级放大。AWS SNS 和 SQS 正是为了解决这类问题而存在的消息组件SNS 负责把事件广播给多个订阅方SQS 负责把消息积压、削峰并等待消费者按需处理。本系列第一篇会从概念讲起用一个最小可运行案例把“SNS 发布事件 - SQS 接收事件 - 消费者处理事件”的完整链路跑通随后解释可靠性设计、死信队列和常见排查思路。读完这篇后你能在自己的微服务项目里判断哪类场景该用 SNS、哪类场景该用 SQS也能独立搭出一条事件驱动链路。1. 为什么微服务架构需要消息中间件而不是只靠 HTTP 同步调用1.1 同步调用在微服务链路中的四个问题微服务之间的同步调用不是不能使用而是需要认识到它的代价。假设下单接口内部要依次调用 4 个下游服务整个接口的响应时间就是 4 个下游响应时间之和再加上网关、序列化和网络传输时间。任何一个下游服务只要出现 300 毫秒延迟下单接口就会从原本的 50 毫秒变成 350 毫秒。这个延迟叠加问题在调用链变深后会非常明显。更麻烦的是可用性问题。4 个下游服务只要有一个不可用下单主流程可能直接失败或超时。即使服务之间配置了超时时间和重试机制同步重试也会把流量继续压向下游。当一个服务宕机依赖它的所有上游都会因为重试而增加积压请求最终形成雪崩效应。还有一类情况是业务上根本不需要立即返回结果。订单创建成功后系统要发送邮件、生成推荐数据、更新报表、推送消息给仓配系统。用户并不需要等待这些动作全部完成才能看到“下单成功”的响应。如果把这些动作放进同步链等于把非核心路径的延迟强加给了核心路径。第四个问题是流量峰值。电商大促、秒杀、活动抢购都会出现短时间的高并发。下游服务如果按照峰值流量扩容成本很高如果不扩容又会在高峰期被打垮。消息队列可以在上游和下游之间增加一个缓冲层让瞬时流量先进入队列再由消费者按自己的处理能力把消息消费完。1.2 SNS 和 SQS 的分工一个负责广播一个负责缓冲SNS 的全称是 Simple Notification Service定位是发布/订阅模型。生产者把消息发布到 TopicTopic 会把消息推送给所有订阅者。订阅者可以是 SQS 队列、Lambda 函数、HTTP 端点、邮件地址、移动终端等。SNS 不负责缓存消息给消费者慢慢取它更多是事件分发器。SQS 的全称是 Simple Queue Service定位是消息队列。生产者把消息发送到 Queue消费者主动轮询拉取消息处理成功后删除消息。SQS 会把消息保留一段时间默认保留 4 天最长可以配置成 14 天。消费者不在线时消息会一直留在队列里由后来启动的消费者继续处理。在微服务架构中SNS 和 SQS 的组合非常常见。一个订单创建事件发布到 SNS Topic 后SNS 可以把消息推送给“库存服务的 SQS 队列”“通知服务的 SQS 队列”“数据仓库的 SQS 队列”。每个下游服务只消费自己的队列互不干扰。这样不仅把事件分发和异步处理分开还让下游服务在消费时拥有独立的重试和积压能力。1.3 一条典型的事件驱动链路一个典型的订单场景可以这样设计订单服务在数据库事务提交成功后发布ORDER_CREATED事件到 SNS Topic。SNS 根据订阅关系把事件推送给 N 个 SQS 队列。库存服务从自己的队列中拉取事件扣减库存通知服务从自己的队列拉取事件发送消息分析服务从自己的队列拉取事件写入数据仓库。发送到 SNS 的消息并不会占用订单事务时间。订单服务只需要把消息发布成功就算完成了事件通知。下游服务如果暂时不可用消息会留在 SQS 队列中下游恢复后消费者会继续从队列中拉取并处理。这样整个系统从同步调用变成了异步事件驱动核心链路稳定性不再受非核心服务影响。2. 先把 SNS 和 SQS 的机制弄明白再动手配置2.1 SNS Topic发布/订阅模型背后是推送关系SNS 的核心概念是 Topic。一个 Topic 相当于一个广播频道。生产者把消息 publish 到 TopicTopic 会把消息推送给所有订阅者。订阅者之间是独立的一个订阅者处理失败不会影响另一个订阅者。SNS 的消息结构里包含几个关键字段Type表示消息类型一般是NotificationMessageId是消息唯一标识TopicArn是主题的 ARNMessage是业务负载Timestamp是发布时间如果需要还可以包含MessageAttributes用于描述消息格式、消息标签等。订阅方收到的就是一个 JSON 格式的消息壳。这里容易误解的点是SNS 不会等待订阅者消费成功后再返回。SNS 的发布 API 只需要确认消息已经投递给订阅通道比如已经交给 SQS 队列就算成功。至于 SQS 里的消息何时被消费者处理SNS 不关心。2.2 SQS Queue为什么消费者要主动拉取而不是被动接收SQS 和常见的消息队列中间件不同它默认采用拉模型。消费者调用ReceiveMessage从队列里取一批消息取到之后这些消息会进入不可见状态其他消费者在这个不可见期间无法取到。消息处理成功后消费者调用DeleteMessage删除该消息如果处理失败消费者可以不删除消息会在可见性超时结束后重新出现在队列里。这个机制保证了消息不会被多个消费者同时拉取也允许消费者根据自身负载能力控制消费速率。消费者如果处理速度不够队列中的消息会积压但不会丢失。通过配置长轮询消费者可以设置WaitTimeSeconds让 SQS 在没有消息时等待最长 20 秒再返回空响应从而减少空轮询带来的 API 请求数量。SQS 默认提供标准队列它的语义是至少一次投递也就是说消息可能被重复投递。另一个类型是 FIFO 队列它提供严格顺序和恰好一次语义但每秒吞吐量有限制。对于大多数微服务事件转发场景标准队列就够用但业务处理逻辑必须设计成支持重复消息。2.3 SNS 和 SQS 的选型对照在实际项目中SNS 和 SQS 不是二选一的关系。它们解决的问题不同经常叠加使用。下面这张表格可以帮助快速判断场景。场景推荐方案原因一个订单事件要发给库存、通知、报表等多个下游SNS Topic 多个 SQS QueueSNS 负责分发SQS 为每个下游提供独立缓冲只有一个消费者消息要异步慢慢处理SQS 单队列使用队列缓存流量峰值消费者按需拉取需要推送给移动设备或邮件SNS 直接订阅移动端或邮箱SNS 原生支持多种推送协议需要严格顺序处理某个业务实体的消息SQS FIFO 队列FIFO 保证同 GroupId 内的消息顺序需要事件广播给 Lambda 做实时处理SNS 订阅 LambdaLambda 被触发后可以直接处理事件某条消息需要延迟一段时间后再处理SQS DelaySeconds 参数生产者发送时或队列配置中设置延迟时间这里要特别说明一个 SNS Topic 可以直接订阅 SQS 队列也可以订阅多个 SQS 队列。这是微服务事件驱动中最常见的组合方式。不要在架构设计里把 SNS 和 SQS 当成互相替代品它们组合起来才能兼顾广播和缓冲。2.4 为什么 SNS 直接订阅 SQS 比“业务服务自己轮询 SNS”合适有些第一次接触事件驱动的同学会问能不能只用一个 SNS Topic让每个服务自己轮询 SNS这里有个概念误区。SNS 不是“消息仓库”它不会保存消息供消费者随时来取消息发布后只负责推送给订阅端。如果某个订阅端不在线SNS 对 HTTP 这类协议会有重试策略但不会长期存储消息。SQS 才是承担消息存储和积压的组件。所以正确的事件驱动架构是SNS 负责“把事件告诉所有感兴趣的订单方”SQS 负责“把事件安全地保存到订单方自己的队列里”。每个订单方从自己的 SQS 队列拉取事件处理完成后删除消息。这套组合让消息既有广播能力又有持久化缓冲能力。3. 搭建实验环境准备账号、CLI、SNS 主题和 SQS 队列3.1 环境准备与前置知识要跟着完整跑通本文案例你需要准备以下内容一个 AWS 账号一双有权限创建 SNS 和 SQS 资源的 IAM 用户或角色本机安装并配置好 AWS CLI本机安装 Python 3 和 boto3一个用于实验的 Region比如us-east-1。先确认本机 AWS CLI 已经配置成功。aws --version aws sts get-caller-identity正常输出会包含你的账号 ID、ARN 和用户名称。如果返回Unable to locate credentials需要先执行aws configure配置Access Key ID、Secret Access Key、默认 Region 和输出格式。接下来安装 Python 依赖后续消费者脚本会用到 boto3。pip install boto3这里要注意学习环境可以使用权限较大的凭证但生产环境一定要按最小权限原则分配 IAM 策略。文中后续会给生产环境的最小权限示例。3.2 通过 CLI 创建 SQS 队列和 SNS 主题创建实验资源时先定义几个变量。export AWS_REGIONus-east-1 export TOPIC_NAMEorder-events-topic export QUEUE_NAMEorder-events-queue创建 SNS 主题。aws sns create-topic --name $TOPIC_NAME --region $AWS_REGION创建 SQS 队列配置可见性超时和消息保留时间。aws sqs create-queue \ --queue-name $QUEUE_NAME \ --region $AWS_REGION \ --attributes { VisibilityTimeout: 30, MessageRetentionPeriod: 86400, ReceiveMessageWaitTimeSeconds: 20 }参数说明VisibilityTimeout是消息被消费者拉取后的不可见时长这里设置为 30 秒MessageRetentionPeriod是消息最大保留时间这里设置为 86400 秒也就是 1 天ReceiveMessageWaitTimeSeconds是长轮询的等待时间设置成 20 秒可以减少空轮询。查询创建出来的主题 ARN 和队列 ARN。后续配置订阅时需要用到。aws sns list-topics --region $AWS_REGION aws sqs list-queues --region $AWS_REGION aws sqs get-queue-attributes \ --queue-url https://sqs.$AWS_REGION.amazonaws.com/123456789012/$QUEUE_NAME \ --attribute-names QueueArn \ --region $AWS_REGION注意将队列 URL 中的123456789012替换成自己的账号 ID。最简单的做法是先执行aws sqs list-queues拿到完整队列 URL再执行get-queue-attributes。3.3 创建 SNS 到 SQS 的订阅并配置访问策略光有主题和队列还不够需要在两者之间建立订阅关系。SNS 要把消息推送到 SQS必须把 SQS 队列的 ARN 作为订阅端点。aws sns subscribe \ --topic-arn arn:aws:sns:$AWS_REGION:123456789012:$TOPIC_NAME \ --protocol sqs \ --notification-endpoint arn:aws:sqs:$AWS_REGION:123456789012:$QUEUE_NAME \ --region $AWS_REGION订阅命令执行后可以通过下面的命令查看订阅状态。aws sns list-subscriptions-by-topic \ --topic-arn arn:aws:sns:$AWS_REGION:123456789012:$TOPIC_NAME \ --region $AWS_REGION订阅创建初期SubscriptionArn可能是PendingConfirmation当 SNS 成功向 SQS 推送消息后订阅状态会变成已确认。这里的“推送”并不是说 SNS 正在推业务消息而是 SNS 需要验证自己是否具备向 SQS 发送消息的权限。如果队列策略不允许 SNS 发送订阅会一直处于等待确认状态消息也无法进入 SQS。因此需要给 SQS 队列配置一条基于sns.amazonaws.com的发送策略。下面是通过 Python 脚本设置队列策略的方式便于直接写入 JSON。import boto3 import json client boto3.client(sqs, region_nameus-east-1) queue_url https://sqs.us-east-1.amazonaws.com/123456789012/order-events-queue topic_arn arn:aws:sns:us-east-1:123456789012:order-events-topic queue_arn arn:aws:sqs:us-east-1:123456789012:order-events-queue policy { Version: 2012-10-17, Statement: [ { Effect: Allow, Principal: {Service: sns.amazonaws.com}, Action: sqs:SendMessage, Resource: queue_arn, Condition: { ArnEquals: { aws:SourceArn: topic_arn } } } ] } client.set_queue_attributes( QueueUrlqueue_url, Attributes{Policy: json.dumps(policy)} ) print(Queue policy set.)这里使用ArnEquals条件把来源限制到指定 SNS Topic而不是允许所有 SNS 主题都往该队列发消息。这种做法在多个业务共用账号的微服务环境中非常重要否则一个其他业务的 SNS 主题也能往这个队列写消息会导致消息污染。检查订阅确认状态。aws sns get-subscription-attributes \ --subscription-arn arn:aws:sns:us-east-1:123456789012:order-events-topic:你的订阅ID \ --region $AWS_REGION如果SubscriptionArn是完整的 ARN并且PendingConfirmation不再出现说明订阅已经生效。3.4 验证订阅状态的小清单完成这一步后用下面的清单确认所有基础资源都正常。检查项预期结果检查命令SNS Topic 存在返回 TopicArnaws sns list-topicsSQS 队列存在返回 QueueUrlaws sqs list-queues订阅状态已确认SubscriptionArn 不是 PendingConfirmationaws sns list-subscriptions-by-topic队列策略已配置策略中包含 SNS 发送权限aws sqs get-queue-attributes --attribute-names Policy这一节完成后SNS 和 SQS 之间的通道已经打通。下一步就是发布消息并且消费它。4. 把消息流跑通一个订单事件从发布到消费4.1 设计事件结构进入代码之前先约定事件结构。事件是服务之间传递的语义单位推荐包含事件类型、事件ID、业务实体的关键字段和发生时间。{ eventId: evt_10001, eventType: ORDER_CREATED, orderId: order_20260001, amount: 299.00, createdAt: 2026-01-01T10:00:00Z }eventId是事件的唯一标识可以用来做幂等去重eventType是事件的语义类型orderId是业务实体主键amount是订单金额createdAt是事件发生时间。事件结构应尽量只包含消费者做决策所必需的数据不要塞入数据库表结构等内部细节。4.2 发布消息到 SNS Topic发布消息可以使用 CLI也可以使用 Python。先看 CLI 方式。aws sns publish \ --topic-arn arn:aws:sns:us-east-1:123456789012:order-events-topic \ --message {eventId:evt_10001,eventType:ORDER_CREATED,orderId:order_20260001,amount:299.00,createdAt:2026-01-01T10:00:00Z} \ --region $AWS_REGION返回结果中会包含MessageId说明 SNS 已经接收消息。用 Python 发布时代码更接近实际生产项目的写法。import boto3 import json client boto3.client(sns, region_nameus-east-1) response client.publish( TopicArnarn:aws:sns:us-east-1:123456789012:order-events-topic, Messagejson.dumps({ eventId: evt_10001, eventType: ORDER_CREATED, orderId: order_20260001, amount: 299.00, createdAt: 2026-01-01T10:00:00Z }) ) print(response[MessageId])这里 SNS 的Message参数虽然是字符串但业务上通常放入一段 JSON。消费者收到后需要先解析外层 SNS 消息壳再解析内层业务 JSON。4.3 用 Python 消费者从 SQS 拉取消息下面是一个最小可运行的消费者脚本它不断从 SQS 队列中拉取消息解析业务事件然后删除消息。import boto3 import json client boto3.client(sqs, region_nameus-east-1) queue_url https://sqs.us-east-1.amazonaws.com/123456789012/order-events-queue while True: response client.receive_message( QueueUrlqueue_url, MaxNumberOfMessages10, WaitTimeSeconds20, VisibilityTimeout30 ) for message in response.get(Messages, []): # 第一层SNS 推送通知的消息壳 envelope json.loads(message[Body]) # 第二层业务事件 event json.loads(envelope[Message]) print(收到事件类型:, event[eventType]) print(订单ID:, event[orderId]) print(金额:, event[amount]) # 业务处理成功后必须删除消息 client.delete_message( QueueUrlqueue_url, ReceiptHandlemessage[ReceiptHandle] )这段代码有几个关键点。第一次解析message[Body]得到的是 SNS 信封里面包含Type、MessageId、TopicArn、Message等字段。第二次解析envelope[Message]得到的是业务事件 JSON。ReceiptHandle是删除消息的唯一凭证。每次收到消息时都要使用当次返回的ReceiptHandle不能提前存下来长期复用。如果消费者在处理完业务后不调用delete_message这条消息会在可见性超时后再次出现在队列中导致重复消费。这里使用WaitTimeSeconds20的长轮询可以避免消费者在没有消息时频繁向 SQS 发起空请求。短轮询每个请求都会立即返回即使没有消息也会产生 API 费用长轮询能显著减少请求次数。4.4 验证消息链路是否已打通消费者脚本运行后再回到另一个终端发布一条消息。正常情况下消费者终端会输出收到事件类型: ORDER_CREATED 订单ID: order_20260001 金额: 299.0你还可以去 SQS 控制台查看队列的NumberOfMessagesSent和NumberOfMessagesReceived指标确认消息已经被 SQS 接收并且消费者已经拉取成功。如果消费者脚本运行后没有输出先检查订阅状态和队列策略再检查消息是否进入了死信队列或是否被其他消费者取走。在这个最小链路里订单服务就相当于发送方脚本库存服务或通知服务就相当于消费者脚本。两者之间不需要知道彼此的地址和接口只依赖 SNS Topic 和 SQS Queue 的 ARN。这就是事件驱动架构的解耦价值。5. 设计可靠的异步消息链路重试、死信与幂等5.1 SQS 不丢消息但可能重复投递消息SQS 标准队列的语义是至少一次投递。也就是说正常情况下每条消息至少会被消费一次但极端情况下可能重复。这与传统数据库事务的单次提交逻辑不一样不要拿“消息不会重复”的假设去设计业务。重复可能出现在两个地方一是 SQS 内部在极端情况下对同一条消息进行了多次可见性恢复二是消费者处理消息时超过了可见性超时时间消息在业务还没有处理完成时又变回可见被另一个消费者拉取。因此消费者代码不能只做“收到消息就处理、处理完就删除”这一套动作还需要在业务层面做到幂等。所谓幂等就是同一条业务事件即使被执行两次最终结果也和执行一次一样。5.2 消费失败时不删除消息让重试自然发生消费者处理业务时可能出现数据库连接失败、下游接口超时、数据格式不正确等问题。如果处理失败就调用delete_message消息会丢失如果处理不成功也不删除消息会在可见性超时结束后重新回到队列等待下一次消费。推荐的流程是先尝试执行业务逻辑业务成功后删除消息业务失败则不要删除消息让 SQS 在可见性超时后重新投递。这里要结合日志记录失败原因避免消息反复重试但没有任何排查线索。try: event json.loads(envelope[Message]) process_order(event) client.delete_message( QueueUrlqueue_url, ReceiptHandlemessage[ReceiptHandle] ) except Exception as exc: print(消息处理失败:, exc) # 不删除消息等待 SQS 重新投递不要在高频路径上直接打印完整消息内容要打印消息 ID、事件类型、失败原因方便后续定位。5.3 配置死信队列拦截多次重试的坏消息如果某条消息因为数据格式问题一直处理失败它会永远在队列里反复出现占用消费线程也影响队列健康。SQS 提供死信队列机制可以把多次重试仍然失败的消息转移到一个独立队列业务工程师可以从死信队列中单独排查。先创建一个死信队列。aws sqs create-queue \ --queue-name order-events-dlq \ --region $AWS_REGION获取死信队列的 ARN。aws sqs get-queue-attributes \ --queue-url https://sqs.us-east-1.amazonaws.com/123456789012/order-events-dlq \ --attribute-names QueueArn \ --region $AWS_REGION然后给主队列配置 RedrivePolicy。import boto3 import json client boto3.client(sqs, region_nameus-east-1) queue_url https://sqs.us-east-1.amazonaws.com/123456789012/order-events-queue dlq_arn arn:aws:sqs:us-east-1:123456789012:order-events-dlq client.set_queue_attributes( QueueUrlqueue_url, Attributes{ RedrivePolicy: json.dumps({ deadLetterTargetArn: dlq_arn, maxReceiveCount: 3 }) } )maxReceiveCount表示消息最多被消费者拉取 3 次。如果 3 次都失败消息会被转移到死信队列。这样主队列不会一直被坏消息阻塞运维人员可以从死信队列中恢复或修正这些消息。需要注意的是死信队列本身不能再配置指向原队列的 RedrivePolicy否则会形成循环投递。检测死信队列的消费者时通常需要人工介入修复数据后把消息重新投递回主队列或直接把消息归档。5.4 幂等处理在微服务里的落地方式幂等处理不是靠某个框架自动完成的需要业务层自己设计。常见做法是为每条业务事件分配一个唯一事件 ID在消费端记录已经处理过的事件 ID重复消息直接跳过。例如在订单事件里用eventId作为去重键。消费者在处理前先查询去重表或 Redis如果eventId已存在就直接删除消息不再执行业务逻辑。下面是一个伪代码示例。def process_event(event): event_id event[eventId] if redis_client.setnx(fdedup:{event_id}, 1, ex86400): process_order(event) return True return FalseSETNX表示只有键不存在时才写入成功。如果写入成功说明这条事件第一次出现可以执行业务逻辑如果写入失败说明重复消息直接返回。在数据库操作场景里更稳妥的方式是利用数据库唯一约束。比如把订单事件的处理结果表设计成eventId为主键重复插入会触发唯一冲突业务代码捕获冲突后按成功处理。这样即使消费者进程在去重步骤之后、业务提交之前崩溃也能在重启后通过唯一约束拦截重复数据。6. 消息链路常见问题排查与解决6.1 订阅状态是 PendingConfirmation或者一直收不到消息现象执行list-subscriptions-by-topic后SubscriptionArn显示为PendingConfirmation。向 SNS 发布消息后SQS 队列里没有消息。排查顺序检查订阅端点是否写成了 SQS 队列 URL而不是队列 ARN。SNS 订阅 SQS 的notification-endpoint必须是队列 ARN。检查队列策略是否允许 SNS 发送消息。缺少策略时SNS 无法完成订阅确认。检查 Topic 和 Queue 是否在同一个账号、同一个 Region。跨账号也可以配置但需要更复杂的资源策略。检查订阅的RawMessageDelivery属性。如果设置成 trueSQS 收到的消息里不会包含 SNS 信封消费端解析逻辑会不一样。解决方式补上队列策略重新确认订阅状态。对 SQS 订阅一般不需要手动确认订阅只要权限正确订阅 ARN 会自动从 PendingConfirmation 变为完整 ARN。6.2 消费者进程反复收到同一条消息现象消费者明明处理并删除了消息但过一会儿又收到同一条消息或者消息处理时间较长还没有删就已经被重新投递。常见原因有两个。第一个是消息处理时间超过了可见性超时。假设消费者处理一条消息需要 40 秒但队列的VisibilityTimeout只有 30 秒消息在 30 秒后重新可见其他消费者就能再次拉取到它。第二个原因是消费者处理失败后没有删除消息消息在可见性超时后重新回到队列。这种情况下需要看日志确认是业务失败还是删除失败。解决方式把队列的VisibilityTimeout设置成大于消费者单条消息处理时间对于耗时非常长的任务可以在处理过程中调用ChangeMessageVisibility延长不可见时间。6.3 消费者解析消息时报错或字段不存在现象消费端使用event[orderId]时报KeyError或TypeError。原因通常是解析层级不对。SNS 订阅 SQS 后队列中的消息是一个复合结构。常见错误是只解析了一层直接用json.loads(message[Body])[orderId]但订单 ID 在外层信封的Message字段里。解决方式先解析外层得到 SNS 信封再解析envelope[Message]得到业务消息。如果订阅开启了RawMessageDelivery则队列里直接就是 SNS 发布时的原始消息内容不需要再解析信封。两种模式必须保持一致否则代码会解析失败。6.4 从指标到日志的排查顺序消息链路涉及发布端、SNS、SQS、消费端四个环节排查时要按“消息是否发布 - 是否进入队列 - 是否被接收 - 是否被删除”的顺序推进。现象检查位置可能原因SNS 发布失败SNS 发布调用日志、IAM 权限缺少 sns:Publish 权限、TopicArn 拼写错误队列消息数为 0SQS 控制台指标订阅未确认、队列策略错误、发布端没有执行成功队列有消息但消费者不消费消费者日志、可见性超时消费者没有运行、IAM 缺少 ReceiveMessage 权限消息反复出现消费者日志、RedrivePolicy处理失败未删除、可见性超时过短、缺少幂等处理消息进入死信队列死信队列指标maxReceiveCount 设置过低、业务代码存在无法修复的数据问题AWS 控制台的 CloudWatch 指标可以直接看到NumberOfMessagesSent、NumberOfMessagesReceived、NumberOfMessagesDeleted、ApproximateAgeOfOldestMessage。这些指标可以帮助快速定位消息卡在哪个环节。7. 从实验环境走向生产环境还需要补齐哪些能力7.1 学习环境与生产环境的配置差异本文的演示只适合跑通链路直接照搬到生产环境会有明显风险。下面这张表整理了主要差异。事项学习环境生产环境凭证权限可使用管理员权限使用最小权限 IAM 策略区分发布者和消费者角色队列策略允许单 Topic 写入即可限定账号、Topic、Region配置更严格的 SourceArn 条件加密可不配置启用 SQS 加密使用 AWS KMS 托管密钥监控手动查控制台配置 CloudWatch 告警关注队列积压和消费失败错误处理print 日志结构化日志、错误追踪、死信队列人工处理流程消息结构手动维护 JSON尽量使用显式事件契约发布前校验字段幂等可能不做必须有事件 ID 和去重逻辑否则重复消息会造成数据错误生产环境里消费端至少还需要考虑配置外置化、日志采集、监控告警和异常逃逸处理。消费者进程要能被编排工具管理异常退出后自动重启不能再靠终端前台运行脚本。7.2 最小权限 IAM 策略示例发布者角色只需要调用 SNS Publish 的权限并且限制到指定 Topic。{ Version: 2012-10-17, Statement: [ { Effect: Allow, Action: sns:Publish, Resource: arn:aws:sns:us-east-1:123456789012:order-events-topic } ] }消费者角色只需要对指定队列执行接收、删除和修改可见性超时的权限。{ Version: 2012-10-17, Statement: [ { Effect: Allow, Action: [ sqs:ReceiveMessage, sqs:DeleteMessage, sqs:ChangeMessageVisibility ], Resource: arn:aws:sqs:us-east-1:123456789012:order-events-queue } ] }资源策略和身份策略要配合使用。队列策略控制“谁可以往队列发送消息”IAM 策略控制“某个身份可以执行哪些操作”。不要为了省事给开发人员配置对所有 SNS 和 SQS 资源的*权限。7.3 成本控制与监控要点SNS 和 SQS 的计费主要看请求次数和传输流量消息大小对成本也有影响。SQS 单条消息最大为 256 KBAPI 计费以 64 KB 为最小单位计算。消费者如果使用短轮询并且频繁调用ReceiveMessage空请求也会产生费用因此生产代码应优先使用长轮询。监控上重点关注三个指标ApproximateNumberOfMessagesVisible表示队列中可见消息数ApproximateAgeOfOldestMessage表示最老消息的滞留时间NumberOfMessagesReceived和NumberOfMessagesDeleted用于对照消费者处理情况。最老的滞留给消息年龄不断增长时说明消费者处理能力不足或消息一直失败。建议对主队列的死信队列配置告警一旦有消息进入死信队列就触发人工排查流程。这个告警是事件驱动系统里最重要的生产告警之一。7.4 本系列的下一步扩展方向这一篇解决了 SNS 和 SQS 的基础链路Topic、Queue、订阅、发布、消费、死信和幂等。配置好这套基础后你可以在自己的微服务项目里做两类扩展。第一类是把 SNS 消息本身加过滤策略让不同下游只接收自己关心的事件类型减少无效消息。第二类是用 Lambda 作为 SQS 事件的消费者利用 SQS 触发 Lambda 的能力省掉显式运行消费者进程的运维成本。如果业务对消息顺序有严格要求下一步应该研究 SQS FIFO 队列。FIFO 队列支持按MessageGroupId分组同一组内的消息严格有序且不会重复但吞吐量有上限适用范围和标准队列不同。这部分内容适合作为本系列的第二篇主题。在把更多服务接入事件驱动之前最值得投入的仍然是两件事把消息处理设计成幂等把死信消息的排查流程跑通。只有这两个基础点稳了依赖 SNS 和 SQS 的大规模异步链路才有资格进入生产环境。