r2dbc postgres 驱动程序 - 如何从 PooledConnection 中提取 PostgresqlConnection答案

【问题标题】：r2dbc postgres driver - how to extract PostgresqlConnection from PooledConnectionr2dbc postgres 驱动程序 - 如何从 PooledConnection 中提取 PostgresqlConnection
【发布时间】：2021-01-01 07:07:19
【问题描述】：

嗨，我的 Spring Boot 应用程序使用属性文件自动配置了 r2dbc 连接池：

    spring.r2dbc.url=r2dbc:pool:postgres://localhost:5432/ecom
    spring.r2dbc.username=xxx
    spring.r2dbc.password=yyy</code></pre>

现在我需要获取一个 PostgresqlConnection 实例，我这样做：

this.connection = Mono.from(connectionFactory.create()).cast(PostgresqlConnection.class).block();

但是因为这是一个池配置，我收到一个 ClassCastException 和以下包装所需 PostgresqlConnection 的 PooledConnection 对象：

PooledConnection[PostgresqlConnection{client=io.r2dbc.postgresql.client.ReactorNettyClient@14c93774, codecs=io.r2dbc.postgresql.codec.DefaultCodecs@62a68bcb}]

我需要访问 PostgresqlConnection 并使用它的原生功能，例如通知：

PostgresqlConnection connection = …;
    Flux<Notification> listen = connection.createStatement("LISTEN mymessage")
    .execute()
    .flatMap(PostgresqlResult::getRowsUpdated)
    .thenMany(connection.getNotifications());

问题是如何从 connectionFactory 中正确获取 PostgresqlConnection 实例？任何帮助将不胜感激。

【问题讨论】：

标签： postgresql spring-data-r2dbc r2dbc r2dbc-postgresql

【解决方案1】：

覆盖默认的 ConnectionFactory。

 @Bean
 @Primary
 public ConnectionFactory connectionFactory() {
     return new PostgresqlConnectionFactory(
             PostgresqlConnectionConfiguration.builder()
                     .host("localhost")
                     .database("test")
                     .username("user")
                     .password("password")
                     .codecRegistrar(EnumCodec.builder().withEnum("post_status", Post.Status.class).build())
                     .build()
     );
 }

为监听/通知创建另一个连接工厂。

    @Bean
 @Qualifier("pgConnectionFactory")
 public ConnectionFactory pgConnectionFactory() {
     return new PostgresqlConnectionFactory(
             PostgresqlConnectionConfiguration.builder()
                     .host("localhost")
                     .database("test")
                     .username("user")
                     .password("password")
                     //.codecRegistrar(EnumCodec.builder().withEnum("post_status", Post.Status.class).build())
                     .build()
     );
 }

我已经为第二种方法创建了一个示例，请查看here。

启动应用程序，从curl发送你好：

curl http://localhost:8080/hello

在控制台中，您将看到如下消息：

2020-09-15 16:49:20.657  INFO 20216 --- [ctor-http-nio-4] sending notification::                   : onSubscribe(FluxFlatMap.FlatMapMain)
2020-09-15 16:49:20.658  INFO 20216 --- [ctor-http-nio-4] sending notification::                   : request(unbounded)
2020-09-15 16:49:20.666  INFO 20216 --- [actor-tcp-nio-2] reactor.Flux.ConcatMap.2                 : onNext(NotificationResponseWrapper{name=mymessageprocessId=753parameter=Hello world at 2020-09-15T16:49:20.656715600})
2020-09-15 16:49:20.667  INFO 20216 --- [actor-tcp-nio-2] com.example.demo.Listener                : notifications: NotificationResponseWrapper{name=mymessageprocessId=753parameter=Hello world at 2020-09-15T16:49:20.656715600}

【讨论】：