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

问题描述

嗨,我的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实例?任何帮助将不胜感激。

解决方法

  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()
         );
     }
    
  2. 创建另一个用于监听/通知的连接工厂。

        @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}