使用Spring Cloud Stream如何实现Kafka主消费暂停切换其他topic后恢复

该需求完全可以基于Spring Cloud Stream的现有原生能力实现，以下是具体实现方案：

核心实现步骤 #

• 前置依赖配置
  引入spring-boot-starter-actuator依赖，并在配置文件中开启bindings管控端点：

  management:
    endpoints:
      web:
        exposure:
          include: bindings
  

• 配置双消费者绑定
  同时配置两个消费者绑定规则，主消费者默认启动，备用消费者默认关闭不自动消费：

  spring:
    cloud:
      stream:
        bindings:
          input-channel:
            destination: 主Topic名称
            group: 主消费者组
            consumer:
              auto-startup: true
          input56-channel:
            destination: INPUT56-CHANNEL
            group: 备用消费者组
            consumer:
              auto-startup: false
  

• 异常检测与消费者切换逻辑
  消费主Topic消息时捕获数据库访问异常，判定为数据库宕机后，通过BindingsEndpoint端点控制消费者启停：

  @Autowired
  private BindingsEndpoint bindingsEndpoint;
  
  // 检测到数据库宕机时调用
  public void switchToBackupChannel() {
      // 暂停主消费者，已拉取未提交的消息会保留偏移量，恢复后不会丢失
      bindingsEndpoint.changeState("input-channel", BindingsEndpoint.State.STOPPED);
      // 启动备用消费者
      bindingsEndpoint.changeState("input56-channel", BindingsEndpoint.State.STARTED);
  }
  

• 存量消费完成检测与主消费者恢复
  因为备用Topic存量不多，在备用消费者的监听逻辑中添加偏移量对比逻辑：
  1. 每次消费完备用Topic的消息后，查询当前备用消费者组的消费偏移量，和Topic最新的分区偏移量做对比
  2. 当所有分区的消费偏移量都等于最新偏移量时，判定存量消息已全部消费完成
  3. 此时先暂停INPUT56-CHANNEL消费者，再重新启动input-channel主消费者即可
• 兜底容错配置
  可以新增一个定时检测任务，每隔固定时间检测数据库可用性，如果数据库已恢复且备用Topic消费完成，自动触发主消费者恢复逻辑，避免异常场景下流程卡住。

注意事项 #

• 主消费者暂停后不会丢失消息，只要配置了持久化消费者组，重启后会从暂停时的偏移量继续消费
• 备用消费者建议配置为消费成功后自动提交偏移量，避免存量消息重复消费

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