如何在Spring Webflux的Reactor-Netty服务器中监控指标

我是Spring Boot和Spring Webflux的新手，正在开发基于Reactor-Netty的Spring Webflux服务器以处理WebSocket连接，服务器核心代码如下：

@Component
public class ServerWebSocketHandler implements WebSocketHandler {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    public Mono<Void> handle(WebSocketSession session) {
        String sessionId = session.getId();
        Sinks.Many<String> unicastSink = Sinks.many().unicast().onBackpressureError();
        // save the unicastSink in cache so that on demand messages can be sent to the sink
        Mono<Void> receiver =
                session
                    .receive()
                    .map(WebSocketMessage::getPayloadAsText)
                    .doOnNext(message -> this.handleIncomingMessage(sessionId, message))
                    .doOnError(error -> {
                        logger.info("Error occurred in the session - Session: '{}'; Error: '{}'", sessionId, error);
                    })
                    .doFinally(s -> {
                        this.cleanUp(sessionId, s);
                    })
                    .then();

        Mono<Void> sender =
                session
                    .send(unicastSink.asFlux().map(session::textMessage));

        return Mono.zip(receiver, sender).then();
    }
    // handleIncomingMessage, cleanUp, and other private methods to handle business logic
}

我希望监控可识别背压或内存泄漏的指标，如reactor.netty.eventloop.pending.tasks、reactor.netty.bytebuf.allocator.used.direct.memory、reactor.netty.bytebuf.allocator.used.heap.memory。我在Reactor Netty参考指南中了解到这些指标，但示例是在创建服务器时启用的，而Webflux中相关逻辑已被抽象。请问在此场景下如何启用这些指标监控并消费指标？若能提供示例代码将十分感谢。

在Spring Boot + WebFlux场景下，启用Reactor Netty指标无需手动操作服务器实例，通过依赖配置和Spring Boot自动装配即可实现，具体步骤如下：

1. 添加监控依赖 #

确保项目依赖中包含Spring Boot Actuator和Micrometer相关组件，这是实现指标收集与暴露的核心：

Maven配置 #

<dependencies>
    <!-- Spring Boot WebFlux核心依赖 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
    <!-- Actuator：暴露监控端点 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>
    <!-- Micrometer：指标收集核心 -->
    <dependency>
        <groupId>io.micrometer</groupId>
        <artifactId>micrometer-core</artifactId>
    </dependency>
    <!-- 可选：对接Prometheus，用于将指标推送到监控系统 -->
    <dependency>
        <groupId>io.micrometer</groupId>
        <artifactId>micrometer-registry-prometheus</artifactId>
    </dependency>
</dependencies>

Gradle配置 #

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-webflux'
    implementation 'org.springframework.boot:spring-boot-starter-actuator'
    implementation 'io.micrometer:micrometer-core'
    // 可选：Prometheus对接依赖
    implementation 'io.micrometer:micrometer-registry-prometheus'
}

2. 启用Reactor Netty指标 #

在application.yml或application.properties中开启Actuator端点，并启用Reactor Netty指标收集：

application.yml示例 #

management:
  endpoints:
    web:
      exposure:
        include: metrics, prometheus  # 暴露metrics基础端点和Prometheus专属端点
  metrics:
    reactor:
      netty:
        enabled: true  # 开启Reactor Netty指标自动收集

application.properties示例 #

management.endpoints.web.exposure.include=metrics,prometheus
management.metrics.reactor.netty.enabled=true

3. 验证指标可用性 #

启动应用后，访问以下端点验证指标是否生效：

• 查看所有可用指标：http://localhost:8080/actuator/metrics，列表中会包含你需要的三个Reactor Netty指标
• 查看单个指标详情：比如访问http://localhost:8080/actuator/metrics/reactor.netty.eventloop.pending.tasks，可获取该指标的维度数据（如不同EventLoop的待处理任务数）

4. 自定义消费指标（可选） #

如果需要在代码中直接获取并处理这些指标数据，可以通过注入MeterRegistry来查询：

import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.search.Search;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

@Component
public class ReactorNettyMetricsMonitor {

    private final MeterRegistry meterRegistry;

    public ReactorNettyMetricsMonitor(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
    }

    @PostConstruct
    public void initMetricsMonitoring() {
        // 查询待处理EventLoop任务数
        Search.in(meterRegistry)
                .name("reactor.netty.eventloop.pending.tasks")
                .meters()
                .forEach(meter -> {
                    System.out.printf("EventLoop [%s] 待处理任务数: %.2f%n",
                            meter.getId().getTags(), meter.measure().get(0).getValue());
                });

        // 查询直接内存使用量
        Search.in(meterRegistry)
                .name("reactor.netty.bytebuf.allocator.used.direct.memory")
                .meters()
                .forEach(meter -> {
                    System.out.printf("ByteBuf分配器直接内存使用量 [%s]: %.2f bytes%n",
                            meter.getId().getTags(), meter.measure().get(0).getValue());
                });

        // 查询堆内存使用量
        Search.in(meterRegistry)
                .name("reactor.netty.bytebuf.allocator.used.heap.memory")
                .meters()
                .forEach(meter -> {
                    System.out.printf("ByteBuf分配器堆内存使用量 [%s]: %.2f bytes%n",
                            meter.getId().getTags(), meter.measure().get(0).getValue());
                });
    }
}

关键提示 #

• Spring Boot自动完成Reactor Netty与Micrometer的绑定，无需手动调用Reactor Netty的Metrics类配置指标
• reactor.netty.eventloop.pending.tasks指标可直接反映EventLoop任务积压情况，是排查背压或线程阻塞的关键依据
• 内存相关指标可以帮助你监控ByteBuf分配器的内存使用趋势，提前发现内存泄漏或OOM风险

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