关闭来自 R2DBC 池的连接 直接使用 R2DBC，带反应器在 jOOQ 中使用 R2DBC

如何解决关闭来自 R2DBC 池的连接 直接使用 R2DBC，带反应器在 jOOQ 中使用 R2DBC

我正在尝试使用连接池配置到 PostgreSQL 数据库的 R2DBC 连接。我已遵循此文档：https://github.com/r2dbc/r2dbc-pool/blob/main/README.md。

ConnectionFactory connectionFactory = ConnectionFactories.get(ConnectionFactoryOptions.builder()
        .option(DRIVER,"pool")
        .option(PROTOCOL,"postgresql") // driver identifier,PROTOCOL is delegated as DRIVER by the pool.
        .option(HOST,"192.168.1.200")
        .option(PORT,5433)
        .option(DATABASE,"XXX")
        .option(USER,"XXX")
        .option(PASSWORD,"XXX")
        .build());

ConnectionPoolConfiguration configuration = ConnectionPoolConfiguration.builder(connectionFactory)
        .maxIdleTime(Duration.ofMillis(1000))
        .initialSize(10)
        .maxSize(100)
        .build();
ConnectionPool pool = new ConnectionPool(configuration);

Mono<Connection> connection = pool.create();
return Flux.from(connection)
        .flatMap(conn -> conn
                .createStatement("SELECT id,name,description FROM assortment")
                .execute())
        .flatMap(result -> result
                .map((row,rowMetadata) ->
                        new AssortmentItem(
                                row.get("id",Long.class),row.get("name",String.class),row.get("description",String.class))
                ));

但我不知道如何在查询后关闭连接，以将其恢复到池中。 我试过了：

connection.close();

但它不起作用，尝试使用资源。在 10 次调用后没有这个，我得到了“io.r2dbc.postgresql.ExceptionFactory$PostgresqlNonTransientResourceException”和消息“对不起，已经有太多客户端”。

在我的示例中如何关闭连接并将其恢复到池中？ 当最大池大小为 100 时，为什么我只能从池中调用 10 个查询（初始池大小）？

解决方法

直接使用 R2DBC，带反应器

这是一种使用 reactor 实现的方法，其灵感来自于 OracleTestKit：

pool.create()
    .flatMapMany(connection ->
        Flux.from(connection.createStatement("SELECT ...").execute())
            .flatMap(it -> it.map((r,m) -> new AssortmentItem(...)))
    )
    .concatWith(Mono.from(connection.close()).cast(AssortmentItem.class))

其他示例见in this discussion on the mailing list。归根结底，这不是关于 R2DBC，而是关于使用您的反应流 API（例如反应器）来正确地对 execute() 和 close() 操作进行排序。

在 jOOQ 中使用 R2DBC

A number of third party APIs 已经在包装 R2DBC。 jOOQ 3.15 will,too。这将使这项特定任务变得更容易：

// Inject this into your beans,or whatever
DSLContext ctx = DSL.using(pool);

// jOOQ queries are publishers
Flux.from(ctx.resultQuery("SELECT id,name,description FROM assortment"))

// Use jOOQ's mapping to automatically map fields by reflection
    .map(r -> r.into(AssortItem.class));

这不是使用或依赖于 jOOQ 的 DSL（尽管您也可以这样做，如果您愿意）。我想 JDBI 和其他 API 会提供类似的简化吗？

免责声明：我为 jOOQ 背后的公司工作。