如何实现RabbitMQ ConsumerC在ConsumerA、B处理完消息后执行？

以下是几种适配现有架构的可行方案，均基于你当前使用的MongoDB和RabbitMQ技术栈：

方案一：MongoDB状态追踪+批次检查触发 #

利用MongoDB记录每个ID的处理状态，在ConsumerA、B处理完批次后检查所有ID是否都完成A、B的处理，满足条件则触发ConsumerC。

步骤1：扩展MongoDB实体类 #

给MongoRecord新增两个布尔状态字段，标记是否被A、B处理完成：

class MongoRecord {
    ObjectId id
    String propertyA
    String propertyB
    String propertyC
    // 新增状态字段
    boolean processedByA = false
    boolean processedByB = false
}

步骤2：修改ConsumerA、B的处理逻辑 #

处理完每个ID后更新对应状态，批次处理结束后检查整个批次的完成状态：

// ConsumerA.groovy 修改后
void handleMessage(List<MongoId> mongoIds) {
  for (MongoId mongoId : mongoIds) {
    MongoRecord mongoRecord = mongoRepository.findById(mongoId)
    // 原有业务逻辑
    mongoRecord.propertyA = client.callExternalService()
    mongoRecord.processedByA = true // 标记A处理完成
    mongoRepository.save(mongoRecord)
  }
  // 检查批次是否可触发ConsumerC
  checkAndTriggerConsumerC(mongoIds)
}

private void checkAndTriggerConsumerC(List<MongoId> mongoIds) {
  // 查询批次中未同时完成A、B处理的ID数量
  long unprocessedCount = mongoRepository.countByMongoIdInAndProcessedByAIsTrueAndProcessedByBIsFalse(mongoIds)
  unprocessedCount += mongoRepository.countByMongoIdInAndProcessedByAIsFalseAndProcessedByBIsTrue(mongoIds)
  unprocessedCount += mongoRepository.countByMongoIdInAndProcessedByAIsFalseAndProcessedByBIsFalse(mongoIds)
  
  if (unprocessedCount == 0) {
    rabbitTemplate.convertAndSend('consumer.c', mongoIds)
  }
}

ConsumerB做完全相同的修改，处理完批次后调用checkAndTriggerConsumerC方法即可。

优缺点 #

• 优点：无需引入额外组件，直接复用现有MongoDB；能精准跟踪单个ID的处理状态，避免批次中部分ID未完成就触发C。
• 缺点：批次检查会增加MongoDB查询开销；需维护实体类的状态字段。

方案二：Redis分布式计数器+批次级触发 #

通过Redis计数器记录ConsumerA、B的批次完成情况，当两个消费者都处理完同一批次时，触发ConsumerC。

步骤1：前置流程新增批次ID与计数器初始化 #

前置发送消息到A、B时，生成唯一批次ID并初始化Redis计数器（值为2，对应两个消费者）：

// 前置流程代码
String batchId = UUID.randomUUID().toString()
// 封装批次ID与mongoIds发送给A、B
BatchMessage batchMsg = new BatchMessage(batchId: batchId, mongoIds: mongoIds)
rabbitTemplate.convertAndSend('consumer.a', batchMsg)
rabbitTemplate.convertAndSend('consumer.b', batchMsg)
// 初始化Redis计数器，设置1小时过期避免内存泄漏
redisTemplate.opsForValue().set(batchId, 2, 1, TimeUnit.HOURS)

步骤2：修改ConsumerA、B处理逻辑 #

处理完批次后递减计数器，当计数器归0时触发ConsumerC：

// ConsumerA.groovy 修改后
void handleMessage(BatchMessage batchMsg) {
  List<MongoId> mongoIds = batchMsg.mongoIds
  // 原有业务逻辑...
  
  // 递减计数器并检查是否归0
  long remaining = redisTemplate.opsForValue().decrement(batchMsg.batchId)
  if (remaining == 0) {
    rabbitTemplate.convertAndSend('consumer.c', mongoIds)
    redisTemplate.delete(batchMsg.batchId) // 清理计数器
  }
}

ConsumerB做完全相同的修改即可。

优缺点 #

• 优点：批次级触发性能高，数据库开销小；适合不关注单个ID状态、只要求整个批次完成的场景。
• 缺点：需引入Redis组件；需处理消费者异常失败导致计数器无法归0的情况（可配合Redis过期时间或死信队列兜底）。

方案三：RabbitMQ延迟队列+状态检查 #

利用RabbitMQ延迟队列实现“延迟检查-触发”逻辑，ConsumerA、B处理完批次后发送延迟消息，延迟队列的消费者检查状态，满足条件则转发给ConsumerC，否则重新延迟检查。

步骤1：配置延迟队列（Spring AMQP示例） #

配置带死信功能的延迟队列，支持消息重新入队延迟：

@Bean
Queue checkQueue() {
    return QueueBuilder.durable("consumer.c.check")
            .withArgument("x-dead-letter-exchange", "")
            .withArgument("x-dead-letter-routing-key", "consumer.c.check")
            .build();
}

@Bean
DirectExchange checkExchange() {
    return new DirectExchange("check.exchange");
}

@Bean
Binding checkBinding() {
    return BindingBuilder.bind(checkQueue()).to(checkExchange()).with("consumer.c.check");
}

步骤2：修改ConsumerA、B发送延迟检查消息 #

处理完批次后发送延迟消息到检查队列：

// ConsumerA.groovy 修改后
void handleMessage(List<MongoId> mongoIds) {
  // 原有业务逻辑...
  
  // 发送5秒延迟的检查消息
  Message message = MessageBuilder.withBody(objectMapper.writeValueAsBytes(mongoIds))
          .setHeader("x-delay", 5000)
          .build();
  rabbitTemplate.send("check.exchange", "consumer.c.check", message);
}

ConsumerB做完全相同的修改即可。

步骤3：实现检查队列消费者 #

从延迟队列取消息，检查状态后决定是否触发ConsumerC：

@Component
class CheckConsumer {
    @Autowired
    MongoRepository mongoRepository
    @Autowired
    RabbitTemplate rabbitTemplate
    @Autowired
    ObjectMapper objectMapper

    @RabbitListener(queues = "consumer.c.check")
    void handleMessage(byte[] messageBody) {
        List<MongoId> mongoIds = objectMapper.readValue(messageBody, new TypeReference<List<MongoId>>() {})
        
        // 检查批次是否所有ID都完成A、B处理
        long unprocessedCount = mongoRepository.countByMongoIdInAndProcessedByAIsFalseOrProcessedByBIsFalse(mongoIds)
        if (unprocessedCount == 0) {
            rabbitTemplate.convertAndSend('consumer.c', mongoIds)
        } else {
            // 未完成则重新发送延迟消息，再等5秒检查
            Message message = MessageBuilder.withBody(messageBody)
                    .setHeader("x-delay", 5000)
                    .build();
            rabbitTemplate.send("check.exchange", "consumer.c.check", message);
        }
    }
}

优缺点 #

• 优点：无需额外数据库频繁查询，延迟检查可适配异步处理的不确定性；复用RabbitMQ组件。
• 缺点：延迟时间需根据业务处理耗时合理设置；可能出现多次重复检查的情况。

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