连接池 Metrics
R2DBC 连接池自带 metrics 快照，不需要自己包装 ConnectionFactory：
import io.r2dbc.pool.ConnectionPool;
import io.r2dbc.pool.ConnectionPoolConfiguration;
import io.r2dbc.spi.ConnectionFactory;
@Configuration
public class R2dbcMetricsConfig {
@Bean
public ConnectionPool connectionPool(ConnectionFactory factory) {
ConnectionPoolConfiguration config = ConnectionPoolConfiguration.builder(factory)
.initialSize(5)
.maxSize(20)
.maxIdleTime(Duration.ofMinutes(10))
.build();
return new ConnectionPool(config);
}
}
这段代码假设底层 ConnectionFactory 由你手工提供、还没有池化。在标准 Spring Boot 应用里，配置了 spring.r2dbc.url 后 Boot 已经创建了一个 ConnectionPool 并作为 ConnectionFactory Bean 暴露；此时再声明 ConnectionPool Bean，Boot 的池化和非池化自动配置都会因为 @ConditionalOnMissingBean(ConnectionFactory.class) 退让，而你的 Bean 又在等一个 ConnectionFactory，结果是找不到可注入的工厂（或形成循环依赖）。Boot 应用的正确做法是配置 spring.r2dbc.pool.*；需要更细的自定义时用 ConnectionFactoryOptionsBuilderCustomizer（调整连接选项）或 ConnectionFactoryDecorator（在 Boot 建好的工厂外面做装饰），不要手工包第二层池。R2dbcCustomizer 在 Boot 3.5 已不存在，不要照旧版本示例引用它。
ConnectionPool 暴露关键指标：
ConnectionPool pool = ...;
int active = pool.getMetrics().map(m -> m.acquiredSize()).orElse(0);
int idle = pool.getMetrics().map(m -> m.idleSize()).orElse(0);
int pending = pool.getMetrics().map(m -> m.pendingAcquireSize()).orElse(0);
int max = pool.getMetrics().map(m -> m.getMaxAllocatedSize()).orElse(0);
System.out.println("Active: " + active + ", Idle: " + idle +
", Pending: " + pending + ", Max: " + max);
| 指标 | 含义 | 关注点 |
|---|---|---|
acquiredSize | 当前借出的连接数 | 长时间贴着 maxAllocatedSize 说明池不够或连接没归还 |
idleSize | 池中空闲连接数 | 长期为 0 且 pending > 0 才是压力信号 |
pendingAcquireSize | 等待连接的请求数 | 持续大于 0 说明池满，请求在排队 |
maxAllocatedSize | 池允许的最大连接数 | 固定配置，用作分母 |
自定义健康指标
Spring Boot Actuator 检测到 ConnectionFactory 后，已经通过 ConnectionFactoryHealthContributorAutoConfiguration 自动注册了一个名为 r2dbc 的健康贡献者，它会借一条连接做校验。下面的自定义实现用于覆盖它或补充池指标明细；注意不要让两个 Bean 同时注册成同名健康组件，必要时用 @ConditionalOnMissingBean 控制，或者只定制自动配置提供的 details。
通过 Spring Boot Actuator 暴露连接池健康状态：
import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.ReactiveHealthIndicator;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;
import io.r2dbc.pool.ConnectionPool;
@Component
public class R2dbcHealthIndicator implements ReactiveHealthIndicator {
private final ConnectionPool pool;
public R2dbcHealthIndicator(ConnectionPool pool) {
this.pool = pool;
}
@Override
public Mono<Health> health() {
return Mono.usingWhen(
Mono.from(pool.create()),
conn -> Mono.just(conn.isOpen()),
conn -> Mono.from(conn.close()))
.map(valid -> {
if (valid) {
int idle = pool.getMetrics().map(m -> m.idleSize()).orElse(0);
int pending = pool.getMetrics().map(m -> m.pendingAcquireSize()).orElse(0);
return Health.up()
.withDetail("pool.idle", idle)
.withDetail("pool.pending", pending)
.build();
}
return Health.down().withDetail("error", "Connection not open").build();
})
.onErrorResume(err ->
Mono.just(Health.down()
.withDetail("error", err.getMessage())
.build()));
}
}
访问 GET /actuator/health：
{
"status": "UP",
"components": {
"r2dbc": {
"status": "UP",
"details": {
"pool.idle": 3,
"pool.pending": 0
}
}
}
}
连接泄漏检测
连接借了没还（如异常导致 conn.close() 未被订阅）会造成泄漏。连接池没有专门的泄漏检测开关，只能靠超时和存活时间把问题变成可观测的失败：
ConnectionPoolConfiguration.builder(factory)
.maxSize(20)
.maxAcquireTime(Duration.ofSeconds(5)) // 获取连接最多等 5 秒
.maxCreateConnectionTime(Duration.ofSeconds(3)) // 创建连接最多 3 秒
.maxLifeTime(Duration.ofMinutes(30)) // 连接最大存活时间
.maxIdleTime(Duration.ofMinutes(10)) // 空闲超时驱逐
.validationQuery("SELECT 1") // 取连接时校验可用性
.build();
池满后，后续请求等待超过 maxAcquireTime 会抛 io.r2dbc.spi.R2dbcTimeoutException（不是 PoolAcquireTimeoutException，那个类不在 r2dbc-pool 里），而不是无限挂起。maxLifeTime 和 maxIdleTime 只能回收连接，不能证明它曾被泄漏；要定位泄漏，得在借出/归还处加日志或用 usingWhen 统一管理生命周期。
自动重连与故障恢复
数据库临时故障（网络抖动、重启）时，应用应自动恢复：
@Bean
public ConnectionPool resilientPool(ConnectionFactory factory) {
return new ConnectionPool(ConnectionPoolConfiguration.builder(factory)
.maxSize(20)
.initialSize(5)
.maxAcquireTime(Duration.ofSeconds(10))
.backgroundEvictionInterval(Duration.ofMinutes(1)) // 后台执行空闲/过期驱逐
.validationQuery("SELECT 1")
.build());
}
backgroundEvictionInterval 只调度空闲和过期连接的驱逐，不做健康探测；判断连接是否真的可用要靠 validationQuery 或 validationDepth。如果启用 validationQuery，每次取连接都会先跑一次轻量查询，代价要算进延迟。
在应用层添加重试：
@Service
public class OrderService {
public Mono<Order> findById(Long id) {
return orderRepo.findById(id)
.retryWhen(Retry.backoff(3, Duration.ofMillis(100))
.filter(err -> err instanceof R2dbcException));
}
}
注意：重试读操作安全，重试写操作需要幂等性保障（第 2.3 节）。
故障降级
数据库彻底不可用时，应用不能无限挂起。配置降级策略：
@Service
public class OrderService {
private final OrderRepository repo;
private final Cache<String, Order> cache; // Caffeine Async（第 9 章）
public Mono<Order> findById(Long id) {
return repo.findById(id)
.retryWhen(Retry.max(2)
.filter(err -> err instanceof R2dbcTimeoutException))
.onErrorResume(err -> {
// 从缓存读取（可能不是最新数据）
Order cached = cache.getIfPresent(String.valueOf(id));
if (cached != null) {
return Mono.just(cached);
}
return Mono.error(new ServiceUnavailableException("Database unavailable"));
});
}
}
降级层级：
- 重试：网络 transient 错误，2–3 次重试可能恢复
- 缓存：返回缓存数据（可能 stale，但可用）
- 错误：完全不可用，返回 503，让调用方处理
连接池调优参数速查
ConnectionPoolConfiguration.builder(factory)
.initialSize(5) // 启动时创建 5 个连接
.maxSize(20) // 最大 20 个连接
.maxIdleTime(Duration.ofMinutes(10)) // 空闲 10 分钟驱逐
.maxLifeTime(Duration.ofMinutes(30)) // 最长存活 30 分钟
.maxAcquireTime(Duration.ofSeconds(5)) // 获取连接超时 5 秒
.maxCreateConnectionTime(Duration.ofSeconds(3)) // 新建连接超时 3 秒
.backgroundEvictionInterval(Duration.ofMinutes(1)) // 后台驱逐间隔
.validationQuery("SELECT 1") // 取连接时校验
.maxValidationTime(Duration.ofSeconds(1)) // 校验超时
.registerJmx(true) // 暴露 JMX metrics
.build();
eventd v2 连接池监控
eventd v2 连接池配置：
spring:
r2dbc:
pool:
initial-size: 5
max-size: 20
max-idle-time: 10m
max-life-time: 30m
max-acquire-time: 5s
配合 Prometheus + Grafana 监控时，不需要自己写 MeterBinder：Spring Boot Actuator 的 ConnectionPoolMetricsAutoConfiguration 在检测到 io.r2dbc.pool.ConnectionPool 后会自动注册 r2dbc.pool.acquired、r2dbc.pool.allocated、r2dbc.pool.idle、r2dbc.pool.pending、r2dbc.pool.max.allocated 和 r2dbc.pool.max.pending 这组 Gauge（Prometheus 下是 r2dbc_pool_*）。getMetrics() 的用途是自定义健康端点里取明细、或接 JMX，不是把指标再接一遍到 Micrometer。
小结
| 能力 | 方案 | 注意 |
|---|---|---|
| 池指标 | Boot 自动注册 r2dbc.pool.*；getMetrics() 取明细 | 自定义端点/JMX 时才直接读快照 |
| 健康检查 | ReactiveHealthIndicator | pool.create() 验证连通性 |
| 泄漏兜底 | maxAcquireTime + maxLifeTime | 超时失败，暴露问题而非无限挂起 |
| 可用性校验 | validationQuery / validationDepth | 驱逐本身不做健康探测 |
| 自动恢复 | 重试 + 缓存降级 | 读操作安全，写需幂等 |
| 故障降级 | 缓存 fallback | stale 数据优于无数据 |
第 8 章结束。下一章进入缓存：9.1 多数据源与读写分离 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。