取消后未释放 r2dbc-pool 连接

编程问答 2022-11-24

问题描述

我对 R2DBC 池有一个奇怪的行为：我们碰巧创建了大量线程并将它们发送到 R2DBC 池以获取数据库连接。当池中的所有 R2DBC 连接都在使用中时，我们创建的线程将排队等待空闲连接可用，这在先前使用的连接被释放时发生。如果我们在它们等待空闲连接时取消这些线程，则会发生以下行为：

  • 即使它们被取消，一些线程仍会获得连接并执行其正常的数据库进程
  • 最重要的是：即使所有线程都被取消并且不再处于活动状态，一些连接被获取并且永远不会被释放。

因此，一些连接不会回到空闲状态。它们保持获取状态并阻止后续连接请求获取这些特定连接。连接保持锁定状态，直到我们重新启动服务。

值得一提的是，我们在获取连接时对数据库进行了查询（我们有一个多租户数据库，并在获取连接时使用 SET SCHEMA 来选择正确的租户）。

我制作了一个程序来重现这个问题。

为了测试，我使用带有 maxConnection=2 的池。 多次调用测试方法 (controller.test) 后，池中的一些连接会无限期地保持获取状态（它们应该全部被 onCancel 或 close 释放> Spring 处理的语句）。这可以通过使用 jmx 监控池轻松证明。

我认为取消请求会传播到connectionPool.create()，但是有些迭代似乎有足够的时间在收到取消之前结束预查询，这导致连接对 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 但结果保持不变。

解决方法

如 this article about using jOOQ with 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();

这也被推荐在邮件列表中：

希望将来会在 r2dbc.io 网站上记录：

javajavar2dbcr2dbc-postgresqlspringspringspring-data-r2dbc