Spring Boot(Actuator)处理Couchbase7重平衡节点事件及监听失效问题

可行性说明 #

完全可以实现。Couchbase Java SDK（Spring Data Couchbase的底层依赖）会暴露集群拓扑变更、端点状态变化等一系列事件，结合Spring Boot Actuator的自定义健康指示器，就能在集群重平衡时将应用健康状态设为DOWN。

你的监听器不生效的原因 #

你用Spring的@EventListener监听UnexpectedEndpointDisconnectedEvent无效，是因为这个事件是Couchbase SDK内部事件总线分发的，不属于Spring上下文管理的事件，Spring的事件监听机制无法捕获到它。

完整解决方案 #

步骤1：注册Couchbase SDK原生事件监听器 #

通过Spring注入的Cluster Bean，直接操作Couchbase的EventBus来订阅事件，同时实时检查集群重平衡状态：

import com.couchbase.client.core.cnc.EventBus;
import com.couchbase.client.core.cnc.events.cluster.ClusterTopologyChangedEvent;
import com.couchbase.client.core.cnc.events.endpoint.UnexpectedEndpointDisconnectedEvent;
import com.couchbase.client.java.Cluster;
import com.couchbase.client.java.manager.cluster.ClusterNodeState;
import jakarta.annotation.PostConstruct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

@Component
public class CouchbaseRebalanceMonitor {
    private static final Logger log = LoggerFactory.getLogger(CouchbaseRebalanceMonitor.class);
    private final Cluster couchbaseCluster;
    private boolean isRebalancing = false;

    public CouchbaseRebalanceMonitor(Cluster couchbaseCluster) {
        this.couchbaseCluster = couchbaseCluster;
    }

    @PostConstruct
    public void registerEventListeners() {
        EventBus eventBus = couchbaseCluster.environment().eventBus();
        eventBus.subscribe(event -> {
            // 监听端点意外断开事件
            if (event instanceof UnexpectedEndpointDisconnectedEvent) {
                log.info("捕获Couchbase端点意外断开事件: {}", event.description());
                updateRebalanceStatus();
            }
            // 监听集群拓扑变更（重平衡时一定会触发）
            else if (event instanceof ClusterTopologyChangedEvent topologyEvent) {
                log.info("集群拓扑更新: 节点数={}", topologyEvent.newTopology().nodes().size());
                updateRebalanceStatus();
            }
        });
    }

    private void updateRebalanceStatus() {
        try {
            // 查询所有节点状态，判断是否有节点处于重平衡中
            isRebalancing = couchbaseCluster.nodes().stream()
                    .anyMatch(node -> ClusterNodeState.REBALANCING.equals(node.status().state()));
        } catch (Exception e) {
            log.error("更新重平衡状态失败", e);
            isRebalancing = false;
        }
    }

    // 给健康指示器提供状态查询接口
    public boolean isClusterRebalancing() {
        return isRebalancing;
    }
}

步骤2：实现自定义健康指示器 #

继承AbstractHealthIndicator，根据监控类的状态设置应用健康状态：

import org.springframework.boot.actuate.health.AbstractHealthIndicator;
import org.springframework.boot.actuate.health.Health;
import org.springframework.stereotype.Component;

@Component
public class CouchbaseRebalanceHealthIndicator extends AbstractHealthIndicator {
    private final CouchbaseRebalanceMonitor rebalanceMonitor;

    public CouchbaseRebalanceHealthIndicator(CouchbaseRebalanceMonitor rebalanceMonitor) {
        this.rebalanceMonitor = rebalanceMonitor;
    }

    @Override
    protected void doHealthCheck(Health.Builder builder) {
        if (rebalanceMonitor.isClusterRebalancing()) {
            builder.down()
                    .withDetail("status", "DOWN")
                    .withDetail("message", "Couchbase集群正在执行重平衡操作");
        } else {
            builder.up()
                    .withDetail("status", "UP")
                    .withDetail("message", "Couchbase集群状态正常");
        }
    }
}

关键注意事项 #

• 不要混用Spring事件和Couchbase SDK事件：Couchbase的事件通过自身EventBus分发，必须直接订阅SDK的事件总线才能捕获。
• 单一的UnexpectedEndpointDisconnectedEvent不能作为重平衡的唯一判断依据（网络波动也会触发），结合ClusterTopologyChangedEvent和节点状态查询更准确。
• 可以通过ClusterManager.getClusterInfo()获取更详细的集群元数据，进一步确认重平衡任务的执行状态。

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