Spring Boot通过Actuator实现Kafka健康检查失效，缺失类导入如何解决？

类缺失说明 #

HealthRow、AllStatus均为自定义业务类，Spring官方未提供对应实现，无公共导入路径，需自行编码实现。

1. 前置依赖与配置 #

首先确保项目引入了所需依赖并开启健康检查端点：

• Maven依赖

<!-- Spring Boot Actuator -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<!-- Spring Kafka -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

• 配置文件（application.yml）

management:
  endpoints:
    web:
      exposure:
        include: health
  endpoint:
    health:
      show-details: always

2. 基础版：官方内置Kafka健康检查 #

如果仅需要验证Kafka集群连通性，不需要校验消费者监听器运行状态，无需额外编码：
Spring Boot会自动装配KafkaHealthIndicator，启动后自动加入健康检查聚合结果，逻辑为向Kafka集群发送元数据请求验证连通性，集群异常时健康检查返回DOWN状态。

3. 进阶版：自定义Kafka消费者监听器健康检查 #

如果需要校验每个@KafkaListener对应消费容器的运行状态，按以下步骤实现：

3.1 自定义状态承载类 #

// 单个消费者容器状态承载类，对应参考代码中的HealthRow、ConsumerDetails
@Data
public class ConsumerHealthInfo {
    private String listenerId;
    private String groupId;
    private String topics;
    private boolean isRunning;
    private Status status;
}

// 全局消费者状态聚合类，对应参考代码中的AllStatus
@Data
public class GlobalKafkaConsumerStatus {
    private final String name = "globalKafkaConsumerStatus";
    private int totalListenerCount;
    private int runningListenerCount;
    private Status status;
}

3.2 自定义健康指示器实现 #

@Component
public class KafkaListenerHealthIndicator extends AbstractHealthIndicator {

    private final KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

    public KafkaListenerHealthIndicator(KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry) {
        this.kafkaListenerEndpointRegistry = kafkaListenerEndpointRegistry;
    }

    @Override
    protected void doHealthCheck(Health.Builder builder) {
        Map<String, Object> healthDetails = new TreeMap<>();
        GlobalKafkaConsumerStatus globalStatus = new GlobalKafkaConsumerStatus();
        Collection<MessageListenerContainer> allContainers = kafkaListenerEndpointRegistry.getAllListenerContainers();

        globalStatus.setTotalListenerCount(allContainers.size());
        int runningCount = 0;

        for (MessageListenerContainer container : allContainers) {
            ConsumerHealthInfo info = new ConsumerHealthInfo();
            String listenerId = container.getListenerId();
            info.setListenerId(listenerId);
            // 提取消费配置信息
            if (container.getContainerProperties() != null) {
                info.setGroupId(container.getContainerProperties().getGroupId());
                String[] topics = container.getContainerProperties().getTopics();
                if (topics != null) {
                    info.setTopics(String.join(",", topics));
                }
            }
            // 校验容器运行状态
            boolean isRunning = container.isRunning();
            info.setRunning(isRunning);
            info.setStatus(isRunning ? Status.UP : Status.DOWN);
            if (isRunning) runningCount++;
            healthDetails.put(listenerId, info);
        }

        // 聚合全局状态
        globalStatus.setRunningListenerCount(runningCount);
        globalStatus.setStatus(runningCount == globalStatus.getTotalListenerCount() ? Status.UP : Status.DOWN);
        healthDetails.put(globalStatus.getName(), globalStatus);

        builder.withDetails(healthDetails).status(globalStatus.getStatus());
    }
}

4. 效果验证 #

项目启动后访问http://{服务IP}:{服务端口}/actuator/health即可看到每个Kafka监听器的运行状态和全局聚合结果，任意一个监听器异常停止时，整体健康检查结果会返回DOWN。

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