java - r2dbc-pool 连接取消后未释放
问题描述
我对 R2DBC 池有一个奇怪的行为:我们碰巧创建了大量线程并将它们发送到 R2DBC 池以获取数据库连接。当池中的所有 R2DBC 连接都在使用中时,我们创建的线程排队等待空闲连接可用,这在之前使用的连接被释放时发生。如果我们在等待空闲连接时取消这些线程,则会发生以下行为:
- 即使它们被取消,一些线程也会获得一个连接并通过它们的正常 DB 进程
- 最重要的是:某些连接被获取并且永远不会被释放,即使所有线程都被取消并且不再处于活动状态。
因此,一些连接不会回到空闲状态。它们保持被获取并阻止后续连接请求获取这些特定连接。在我们重新启动服务之前,连接保持锁定状态。
值得一提的是,我们在获取连接时对数据库进行了查询(我们有一个多租户数据库,并在获取连接时使用 SET SCHEMA 来选择正确的租户)。
我做了一个程序来重现这个问题。
为了进行测试,我使用了一个带有maxConnection=2的池。在调用了几次测试方法(controller.test)之后,池中的一些连接仍然被无限期地获取(它们应该都被onCancel或 Spring 处理的close语句释放)。这可以通过使用 jmx 来监控池很容易地证明。
我想取消请求会传播到connectionPool.create(),但是某些迭代似乎有足够的时间在收到取消之前结束 preQuery,这导致连接可供 Spring 使用。在这些情况下,在TestConnectionFactory中看不到取消,并且大约 1/3 次,Spring 不调用connection.close,导致保持获取连接。
@Slf4j
@RestController
public class TestController {
private final TestRepo1 testRepo1;
@Autowired
public TestController(
TestRepo1 testRepo1
) {
this.testRepo1 = testRepo1;
}
@GetMapping("test")
Mono<Void> test(
) {
// Will made 49 queries to the database.
return Mono
.when(
IntStream.range(0, 100)
.mapToObj(i -> Mono.defer(() ->
i == 0 ? // the first element throw an error after 2 seconds, canceling all query not already done.
Mono.just(0)
.delayElement(Duration.ofMillis(2000))
.doOnNext(x -> log.info("{} -> throw", x))
.then(Mono.error(new Exception("FAIL"))) :
testRepo1.query(String.valueOf(i)))
)
.collect(Collectors.toList())
)
.then()
.onErrorResume(e -> Mono.empty()); // avoid propagating error to http response.
}
}
@Slf4j
public class TestConnectionFactory implements ConnectionFactory {
private final ConnectionPool connectionPool;
TestConnectionFactory(ConnectionPool connectionPool) {
this.connectionPool = connectionPool;
}
@Override
public Publisher<? extends Connection> create() {
return createTenantConnection()
.doOnNext(x -> log.info("creation transaction done"))
.doOnCancel(() -> log.info("cancel while creation"));
}
private Mono<Connection> createTenantConnection() {
return connectionPool.create()
.flatMap(connection -> preQuery(connection));
}
private Mono<Connection> preQuery(Connection connection) {
return Mono.from(connection
.createStatement("SELECT 1;") // enough to produce the error, in our real code, this is a SET SCHEMA XXX
.execute())
.doOnCancel(() -> log.info("cancel during preQuery"))
.thenReturn(connection);
}
@Override
public ConnectionFactoryMetadata getMetadata() {
return connectionPool.getMetadata();
}
}
@Configuration
public class MyConfiguration {
@Bean
@Scope("singleton")
ConnectionFactory connectionFactory(
ConnectionPool connectionPool
) {
return new TestConnectionFactory(connectionPool);
}
}
@Slf4j
@Repository
public class TestRepo1 {
// simple query waiting 1 second
private static final String QUERY = "SELECT pg_sleep(1);";
private final DatabaseClient databaseClient;
@Autowired
public TestRepo1(DatabaseClient databaseClient) {
this.databaseClient = databaseClient;
}
public Mono<Void> query(String msg) {
log.info("start query {}", msg);
return databaseClient.execute(QUERY)
.map(row -> "result")
.first()
.doOnCancel(() -> log.info("cancel query {}", msg))
.doOnNext(x -> log.info("query {} result", msg))
.then()
.doOnTerminate(() -> log.info("terminate {}", msg));
}
}
我们将org.springframework.boot 2.3.5.RELEASE与io.r2dbc:r2dbc-postgresql和io.r2dbc:r2dbc-pool 一起使用。
我们尝试升级到io.r2dbc:r2dbc-postgresql 0.8.8.RELEASE和io.r2dbc:r2dbc-pool 0.9.0.M1但结果保持不变。
解决方案
正如这篇关于将 jOOQ 与 R2DBC 一起使用的文章中所解释的,使用 R2DBC管理资源的一种好方法是使用Flux.usingWhen()
,例如
Flux.usingWhen(
pool.create(),
c -> c.createStatement("SELECT col FROM my_table").execute(),
c -> c.close()
)
.flatMap(it -> it.map((r, m) -> r.get(0, String.class)))
.doOnNext(System.out::println)
.subscribe();
这也在邮件列表中被推荐:
- https://groups.google.com/g/r2dbc/c/a7CQAU_u_m0/m/XoRxMPXRBQAJ
- https://groups.google.com/g/r2dbc/c/nyIXQ0EQddQ/m/HQ5rRYTnBQAJ
并有望在未来记录在r2dbc.io网站上:
推荐阅读
- java - 从 excel 读取数据时出错,我试图从头开始设置关键字驱动的框架,未完成但现在出现 wn 错误
- apache-spark-sql - java中的Spark Dataframe sql - 如何转义单引号
- c# - 如何在 SQL Server 的外键中增加 id
- beaker-testing - 安装烧杯服务器
- ios - 如何使用 Eureka 表单从更改文本字段的行中检索值?
- html - 下拉项宽度
- laravel - 调用数组上的成员函数 named()
- java - 对象上的 java.lang.NullPointerException,但我没有使用哪个对象
- mysql - 数据库结构、存储和组织数据
- python - Pandas dataframe.set_index() 删除以前的索引和列