在AWS中实现支持选择性消费的生产者/消费者消息模型方案问询

需求梳理 #

用行李传送带类比这个模型：生产者把"行李"（消息）放到传送带上，消费者自行识别并取走属于自己的"行李"，生产者和传送带（消息通道）无需知晓消息归属，仅由工作人员处理无人认领的消息。核心要求包括：

• 支持多消费者自助拉取目标消息
• 为所有消费者预留足够时间尝试拉取消息
• 理想状态下可追踪所有消费者是否已检查消息，再转入死信队列（DLQ）
• 消息仅能被一个消费者消费

现有AWS服务的局限：

• SQS：不支持服务端选择性拉取，消费者只能盲拉无关消息，易导致非目标消息进入DLQ
• SNS/EventBridge：依赖服务端配置过滤规则，无法让消费者自主选择消息

可行AWS方案 #

方案1：SQS + 客户端过滤 + 消息生命周期管控 #

这是基于现有SQS服务的适配方案，核心是把选择逻辑放到消费者端，同时通过配置保证所有消费者有机会处理消息：

• 生产者侧：发送消息时添加自定义属性（比如consumer-identifier）标记消息归属，同时保留SQS生成的MessageId作为全局唯一标识
• 队列配置：使用标准SQS队列，设置足够长的消息可见性超时（比如根据消费者数量设置为3-6小时），确保所有消费者有机会拉取到消息；同时设置合理的最大接收次数，避免消息过早进入DLQ
• 消费者侧：
  1. 定期拉取队列消息，在本地过滤出与自身标识匹配的消息
  2. 匹配成功：调用DeleteMessage删除消息，完成消费
  3. 匹配失败：调用ChangeMessageVisibility将消息可见性超时重置为初始值，让其他消费者有机会拉取
• 无人认领消息处理：当消息的接收次数耗尽后，自动转入DLQ，由工作人员处理
• 进阶优化（追踪检查状态）：搭配DynamoDB实现消息检查追踪
  • 每条消息生成时，在DynamoDB中创建记录，包含MessageId、已检查消费者列表、消息状态
  • 消费者拉取消息后，先在DynamoDB中标记自身已检查该消息
  • 通过CloudWatch Events触发Lambda定时扫描DynamoDB，当所有消费者都标记检查过且消息未被消费时，主动将消息移入DLQ

方案2：Amazon MQ（RabbitMQ）原生实现 #

Amazon MQ兼容RabbitMQ协议，原生支持消费者自主选择消息的模式：

• 生产者侧：将消息发送到RabbitMQ交换器，设置消息属性或路由键
• 消费者侧：自主创建临时队列，绑定到交换器并设置自定义过滤规则（比如基于消息属性的绑定键），仅接收匹配的消息
• 无人认领消息处理：配置死信交换器（DLX），未被任何消费者匹配的消息会自动转入对应的死信队列，由工作人员处理
• 优势：无需客户端过滤，消息仅路由给目标消费者，天然满足"消息仅被一个消费者消费"的要求

内容的提问来源于stack exchange，提问作者Alexander