Spring R2DBC并行插入PostgreSQL退化为单线程执行问题求助

我尝试用Spring R2DBC向PostgreSQL表并行插入数据，但不管用多连接还是同一连接下多批次/语句并行执行，最终都退化为单线程（reactor-tcp-nio-1）串行执行，性能受严重影响。以下是两种测试场景的代码及日志：

多连接场景 #

@Test
void testMultipleStatementBatch2() {
    final List<List<Tuple2<Integer, String>>> result = Flux.range(0, 100)
            .parallel()
            .runOn(Schedulers.parallel())
            .flatMap(index -> Flux.usingWhen(
                    connectionFactory.create(),
                    connection -> {
                        log.debug("Running index:{}", index);
                        return batch(connection);
                    },
                    Connection::close
            ))
            .doOnNext(tuple2s -> log.debug("result:{}", tuple2s))
            .sequential()
            .collectList()
            .block();
    log.debug("list:{}", gson.toJson(result));
}

private Mono<List<Tuple2<Integer, String>>> batch(Connection connection) {
    log.debug("start running");
    final Batch batch = connection.createBatch();
    IntStream.range(0, 10).boxed()
            .forEach(index ->
                    batch.add("INSERT INTO my_test(key,value,comment) " +
                            "values (" +
                            "'" + index + "'" +
                            ",'value_" + index + "'" +
                            ",'comment_" + index + "'" +
                            ")" +
                            " RETURNING *"));
    return Flux.from(batch.execute())
            .flatMap(result -> result.map((row, rowMetadata) -> {
                log.debug("Result values");
                final Integer id = row.get("key", Integer.class);
                final String value = row.get("value", String.class);
                return Tuples.of(id, value);
            }))
            .collectList();
}

日志片段：

20230831 13:33:31.167 DEBUG [Test worker]          QUERY.debug(249) - Executing query: drop table if exists my_test cascade ; create table my_test(key int,value varchar(255),comment varchar(255))
20230831 13:33:31.171 WARN  [reactor-tcp-nio-1]    ReactorNettyClient.warn(289) - Notice: SEVERITY_LOCALIZED=NOTICE, SEVERITY_NON_LOCALIZED=NOTICE, CODE=00000, MESSAGE=table "my_test" does not exist, skipping, FILE=tablecmds.c, LINE=1186, ROUTINE=DropErrorMsgNonExistent
20230831 13:33:32.256 DEBUG [parallel-6]           StatementConnectionTest.lambda$testMultipleStatementBatch2$3(76) - Running index:1
20230831 13:33:32.257 DEBUG [parallel-6]           StatementConnectionTest.batch(96) - start running
20230831 13:33:32.265 DEBUG [parallel-6]           QUERY.debug(249) - Executing query: INSERT INTO my_test(key,value,comment) values ('0','value_0','comment_0') RETURNING *; INSERT INTO my_test(key,value,comment) values ('1','value_1','comment_1') RETURNING *; ...
...
20230831 13:33:32.312 DEBUG [reactor-tcp-nio-1]    StatementConnectionTest.lambda$batch$7(109) - Result values
...

多批次场景 #

@Test
void runBatchParallelism() {
    var list = Mono.usingWhen(dbService.createConnection(),
                    this::batchParallelism,
                    Connection::close)
            .block();
    log.debug("list:{}", gson.toJson(list));
}

private Mono<List<List<Tuple2<Integer, String>>>> batchParallelism(Connection connection) {
    log.debug("start running");
    return Flux.range(0, 5).parallel().runOn(Schedulers.parallel())
            .flatMap(fluxIndex -> {
                log.debug("flux index:{}", fluxIndex);
                final Batch batch = connection.createBatch();
                IntStream.range(0, 10).boxed()
                        .forEach(index ->
                                batch.add("INSERT INTO my_test(key,value,comment) " +
                                        "values (" +
                                        "'" + index + "'" +
                                        ",'value_" + index + "'" +
                                        ",'comment_" + index + "'" +
                                        ")" +
                                        " RETURNING *"));
                return Flux.from(batch.execute())
                        .flatMap(result -> result.map((row, rowMetadata) -> {
                            log.debug("Result values");
                            final Integer id = row.get("key", Integer.class);
                            final String value = row.get("value", String.class);
                            return Tuples.of(id, value);
                        }))
                        .collectList();
            }).sequential()
            .collectList();
}

日志片段：

20230831 14:05:08.608 DEBUG [Test worker]          StatementConnectionTest.batchParallelism(100) - start running
20230831 14:05:08.646 DEBUG [parallel-3]           StatementConnectionTest.lambda$batchParallelism$9(103) - flux index:4
...
20230831 14:05:08.678 DEBUG [reactor-tcp-nio-1]    StatementConnectionTest.lambda$batchParallelism$7(116) - Result values
...

问题原因 #

1. 单连接串行限制：PostgreSQL单个连接本身是单线程串行的，数据库协议不允许同一连接上同时执行多个操作，即使在Reactor中用parallel()调度，最终还是会被数据库强制串行。
2. 连接池配置不足：多连接场景中，若连接池最大连接数设为1，或配置的连接数无法支撑并行度，所有任务会排队等待同一个连接，导致串行执行。
3. 线程模型特性：reactor-tcp-nio-1是Reactor Netty的IO线程，负责处理数据库网络交互，业务逻辑可以在其他线程执行，但IO操作必然回到该线程，真正并行需要多个连接同时发起请求。

解决办法 #

1. 配置足够的连接池容量 #

使用r2dbc-pool时，调整连接池参数以支持并行：

spring.r2dbc.pool.max-size=10  # 根据并发需求调整，建议和并行度匹配
spring.r2dbc.pool.initial-size=5
spring.r2dbc.pool.max-idle-time=30m

2. 优化多连接并行代码 #

显式指定并行度，使用合适的调度器，确保每个并行任务获取独立连接：

@Test
void testMultipleStatementBatch2() {
    final List<List<Tuple2<Integer, String>>> result = Flux.range(0, 100)
            .parallel(10)  // 并行度匹配连接池max-size
            .runOn(Schedulers.boundedElastic())  // 用boundedElastic避免阻塞IO线程
            .flatMap(index -> Flux.usingWhen(
                    connectionFactory.create(),
                    connection -> {
                        log.debug("Running index:{} on thread:{}", index, Thread.currentThread().getName());
                        return batch(connection);
                    },
                    Connection::close
            ))
            .doOnNext(tuple2s -> log.debug("result count:{}", tuple2s.size()))
            .sequential()
            .collectList()
            .block();
    log.debug("total batches:{}", result.size());
}

3. 同一连接下放弃并行，改用单批量插入 #

单个连接无法并行执行多个Batch，应一次性添加所有插入语句到单个Batch：

private Mono<List<Tuple2<Integer, String>>> singleBatchInsert(Connection connection) {
    final Batch batch = connection.createBatch();
    // 一次性添加所有插入语句，使用参数绑定避免SQL注入
    IntStream.range(0, 50).boxed()
            .forEach(index ->
                    batch.add("INSERT INTO my_test(key,value,comment) values ($1, $2, $3) RETURNING *",
                            index, "value_" + index, "comment_" + index));
    return Flux.from(batch.execute())
            .flatMap(result -> result.map((row, rowMetadata) ->
                    Tuples.of(row.get("key", Integer.class), row.get("value", String.class))))
            .collectList();
}

4. 合理使用调度器 #

避免在IO线程执行业务逻辑，用Schedulers.boundedElastic()处理连接获取和业务逻辑，让IO线程专注于网络交互。

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