Spring Webflux异步PostgreSQL Publisher获取首条结果后停止

从你的描述来看，问题大概率出在postgres-async-driver的使用方式或者RxJava Observable到Reactor Publisher的适配环节，下面分点拆解可能的原因和排查方向：

1. 你可能只是做了一次性查询，而非持续监听新行 #

postgres-async-driver的普通query()调用只会返回当前时刻的结果集，不会自动监听后续插入的行。如果你的逻辑是执行一次SELECT * FROM your_table，那自然只能拿到第一批次的行（甚至可能只有第一行如果结果集处理不当）。

要实现实时流式新行，你需要结合PostgreSQL的LISTEN/NOTIFY机制：

• 第一步：给目标表加一个触发器，当有新行插入时，发送NOTIFY通知到指定频道。比如：

  CREATE TRIGGER notify_new_row
  AFTER INSERT ON your_table
  FOR EACH ROW EXECUTE FUNCTION pg_notify('new_rows_channel', NEW.id::text);
  

• 第二步：用postgres-async-driver的listen()方法订阅这个频道，收到通知后再异步查询新插入的行，然后流式输出：

  PostgresAsyncClient client = PostgresAsyncClient.create("jdbc:postgresql://localhost:5432/db");
  AtomicLong lastProcessedId = new AtomicLong(0);
  
  // 订阅通知频道
  client.listen("new_rows_channel", notification -> {
      // 收到通知后，查询ID大于上次处理的行
      client.query("SELECT * FROM your_table WHERE id > $1", lastProcessedId.get())
            .toObservable()
            .flatMap(result -> Observable.from(result.rows()))
            .subscribe(row -> {
                // 处理行数据并发送到WebSocket
                long currentId = row.getLong("id");
                lastProcessedId.set(currentId);
                // 这里替换成你的WebSocket发送逻辑
                webSocketSession.send(Mono.just(session.textMessage(row.toString()))).subscribe();
            });
  });
  

2. Observable到Reactor Publisher的转换有误 #

WebFlux依赖Reactor的Flux/Mono，而postgres-async-driver返回的是RxJava的Observable。如果转换过程中用了错误的操作符，会导致流被截断：

• 错误示例：用toMono()或者take(1)，只会取第一行：

  // 错误：只拿第一行
  Mono<Row> singleRow = Mono.from(RxReactiveStreams.toPublisher(observable));
  

• 正确做法：用Flux.from()把整个Observable转换成持续的流：

  // 正确：获取所有行（包括后续通过监听拿到的新行）
  Observable<Row> rowObservable = client.query(...)
                                        .toObservable()
                                        .flatMap(result -> Observable.from(result.rows()));
  Flux<Row> rowFlux = Flux.from(RxReactiveStreams.toPublisher(rowObservable));
  
  // 绑定到WebSocket输出
  webSocketSession.send(rowFlux.map(row -> session.textMessage(row.toString()))).subscribe();
  

3. 订阅生命周期或背压问题 #

• 检查你的Observable是否被持续订阅：如果订阅被取消（比如WebSocket连接断开后没有重新订阅），或者因为背压策略不当导致流暂停，都会停止接收后续行。
• 避免在响应式流中使用block()等阻塞操作，这会破坏流的连续性，甚至导致只拿到第一行就终止。

快速排查步骤 #

1. 先单独测试postgres-async-driver的Observable：直接订阅它，插入新行看是否能收到通知和新数据，排除驱动本身的问题。
2. 检查Observable转Publisher的代码，确保用的是Flux而非Mono，没有额外的截断操作符。
3. 确认PostgreSQL的LISTEN/NOTIFY触发器是否正确配置，插入新行时能触发通知。

内容的提问来源于stack exchange，提问作者JJ Zabkar