我试图从postgresql和r2dbc接收异步通知,我使用reactivestream和postgres驱动程序,但我不能声明我是java和reactivestream的新手,我不使用像spring这样的框架。
谢谢你们
package com.poc.r2dbc_listen;
import io.r2dbc.postgresql.*;
import io.r2dbc.spi.*;
import org.reactivestreams.*;
import reactor.core.publisher.Flux;
public class R2dbc_listen {
public static void main(String[] args) {
PostgresqlConnectionFactory connectionFactory = new PostgresqlConnectionFactory(PostgresqlConnectionConfiguration.builder()
.host("127.0.0.1")
.port(5442)
.username("postgres")
.password("psswd")
.database("database")
.build());
Publisher<? extends Connection> connectionPublisher = connectionFactory.create();
Flux<Notification> listen = connectionPublisher.createStatement("LISTEN mymessage")
.execute()
.flatMap(PostgresqlResult::getRowsUpdated)
.thenMany(receiver.getNotifications());
}
}
1条答案
按热度按时间v6ylcynt1#
它的工作原理是: