RabbitMQ发布订阅模式同一消费者多个实例如何防止重复消费？

微服务架构模式中，服务间的通信一般采用HTTP、RPC或者MQ（消息队列）。在这三种方案中，HTTP和RPC是一对一的方式，通常用来进行查询或者命令式的操作，MQ则多用于事件的发布和处理。
在实际项目中我们通常会遇到一种情况：

事件有多个订阅者，有的订阅者部署多个实例，要求每个事件只需要发布一次，每个订阅者都要能收到且仅能有其中一个实例收到并进行处理。

简单说就是既要所有订阅者都能收到消息，又要保证每个订阅者只能消费一次，不能重复消费。那么在使用RabbitMQ作为消息中间件时应该如何处理这个问题？
需求

微服务架构下使用RabbitMQ作为服务总线中间件，有这样一种场景，订单服务在用户提交订单后会发送OrderCreatedEvent事件，需要接收此事件的服务有以下几个：
1、日志服务（单点部署）；
2、消息系统，给用户发送短信、邮件（单点部署）；
3、仓储系统，接收通知备货，扣减库存等（多实例，有负载均衡），同一条消息仅允许一个实例消费；
4、财务系统，记录收入流水（多实例，有负载均衡），同一条消息仅允许一个实例消费；
5、BI系统，做销售统计等；
问题

OrderCreatedEvent发出后，5个服务均可以接收到消息并进行处理，但是仓储和财务部署了多个实例，也就意味着每个实例都能收到相同的消息，这种情况下如何避免重复消费？
方案

• 为OrderCreatedEvent事件建立一个fanout exchange，
  
• 每个需要处理OrderCreatedEvent的订阅者单独建立一个queue，订阅者的所有实例都消费这一个queue的消息。
  
• 将所有OrderCreatedEvent订阅者的queue都bind到第一步建立的exchange上。
  
• 事件发布者将OrderCreatedEvent消息发布到exchange，由exchange将消息路由到对应的queue，最后由queue锁定一个消费者实例并进行投递。
  

以下是示意图
graph LRProducer((订单服务))Exchange{{OrderCreatedEvent交换机（fanout exchange）}}QueueA@{shape: subproc, label: 日志系统队列}QueueB@{shape: subproc, label: 消息系统队列}QueueC@{shape: subproc, label: 仓储系统队列}ConsumerA1([日志系统实例])ConsumerB1([消息系统实例])ConsumerC1([仓储系统实例1])ConsumerC2([仓储系统实例2])ConsumerC3([仓储系统实例3])style Producer fill:#003d5b,color:white,stroke:#003d5bstyle Exchange fill:black,color:white,stroke:blackstyle QueueA fill:#926c15,color:white,stroke:whitestyle QueueB fill:#926c15,color:white,stroke:whitestyle QueueC fill:#926c15,color:white,stroke:whitestyle ConsumerA1 fill:#606c38,color:white,stroke:whitestyle ConsumerB1 fill:#606c38,color:white,stroke:whitestyle ConsumerC1 fill:#666,color:white,stroke:whitestyle ConsumerC2 fill:#606c38,color:white,stroke:whitestyle ConsumerC3 fill:#666,color:white,stroke:whiteProducer --发布---> ExchangeExchange --绑定---> QueueA --消费---> ConsumerA1Exchange --绑定---> QueueB --消费---> ConsumerB1Exchange --绑定---> QueueCQueueC --消费---> ConsumerC1QueueC --消费---> ConsumerC2QueueC --消费---> ConsumerC3四种主要 Exchange 类型

RabbitMQ主要有四种核心的Exchange类型：Fanout（广播）、Direct（精确匹配）、Topic（模式匹配）和Headers（按头匹配），它们决定了消息如何从交换机路由到队列，其中Direct、Topic、Fanout最常用，而Headers和一些插件类型（如x-delayed-message）提供更灵活的路由功能。

• Fanout (扇出)
  
  
  
  • 特点：将消息广播到所有绑定到它的队列。
    
  • 路由规则：忽略路由键 (Routing Key)。
    
  • 场景：适用于日志系统、通知广播等需要消息分发到多个消费者的场景。
    
  
  
• Direct (直连)
  
  
  
  • 特点：消息根据与队列绑定的精确路由键来路由。
    
  • 路由规则：消息的 Routing Key 必须完全匹配 队列绑定的 Routing Key。
    
  • 场景：一个消费者处理特定类型的任务，如处理“用户注册”消息。
    
  
  
• Topic (主题)
  
  
  
  • 特点：使用通配符（* 匹配一个词，# 匹配零个或多个词）进行模式匹配路由。
    
  • 路由规则：消息的 Routing Key 模式需要与 队列绑定的模式 匹配。
    
  • 场景：日志级别过滤，如logs.info.* 匹配所有 info 级别的日志，logs.# 匹配所有日志.
    
  
  
• Headers (头部)
  
  
  
  • 特点：根据消息的 Headers 属性（键值对）进行路由，而不是 Routing Key。
    
  • 路由规则：匹配消息 Headers 中的键值对与绑定时设置的键值对。
    
  • 场景：当需要根据复杂属性匹配时，较少使用，因为性能不如前三者。
    
  
  

Demo

后面我将会提供以上方案的具体实现，包含Java和.net两种版本。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！