Spring Boot R2DBC PostgreSQL批量插入性能优化咨询

你的代码插入3万条数据耗时20分钟，核心原因是没有利用PostgreSQL批量插入的最优机制，以下是针对性的优化建议：

1. 开启PostgreSQL驱动的批量插入重写 #

PostgreSQL的R2DBC驱动默认不会把批量INSERT合并成多值语句，需要开启参数强制重写，在application.yml中添加：

spring:
  r2dbc:
    properties:
      reWriteBatchedInserts: true

这个参数会把多次单条INSERT请求合并成INSERT INTO table (...) VALUES (...), (...), ...的形式，大幅减少网络往返次数。

2. 优化批量插入代码逻辑 #

你当前的循环绑定+statement.add()写法，在未开启上述参数时，本质是执行3万次单条INSERT。可以改成按批次构建多值INSERT，同时控制单批次数据量（比如每1000条一批，避免参数过多触发PostgreSQL限制）：

public Mono<Boolean> insertOrders(List<ClientOrder> orders) {
    if (orders.isEmpty()) {
        return Mono.just(true);
    }
    // 每批次处理1000条，可根据实际调整
    int batchSize = 1000;
    return Flux.fromIterable(Lists.partition(orders, batchSize))
            .flatMap(batch -> {
                // 构建多值INSERT的占位符模板
                String placeholders = IntStream.range(0, batch.size())
                        .mapToObj(i -> String.format("($%d,$%d,$%d,$%d,$%d,$%d,$%d)",
                                i*7+1, i*7+2, i*7+3, i*7+4, i*7+5, i*7+6, i*7+7))
                        .collect(Collectors.joining(","));
                // 替换单值INSERT模板为多值形式
                String sql = OrderQueries.INSERT_ORDERS.replace("VALUES (?)", "VALUES " + placeholders);
                
                // 批量绑定所有参数
                return databaseClient.sql(sql)
                        .bind((statement, index) -> {
                            int orderIndex = index / 7;
                            int paramIndex = index % 7;
                            ClientOrder order = batch.get(orderIndex);
                            switch (paramIndex) {
                                case 0: return statement.bind(index+1, order.getUserId());
                                case 1: return statement.bind(index+1, order.getUserName());
                                case 2: return statement.bind(index+1, order.getCustomerEmail());
                                case 3: return statement.bind(index+1, order.getCustomerMobile());
                                case 4: return statement.bind(index+1, order.getAmount());
                                case 5: return statement.bind(index+1, order.getCreateDate());
                                case 6: return statement.bind(index+1, order.getStatusChangeDate());
                                default: return statement;
                            }
                        })
                        .fetch()
                        .rowsUpdated();
            })
            .reduce(0, Integer::sum)
            .map(totalRows -> totalRows == orders.size());
}

注意：OrderQueries.INSERT_ORDERS需要是单值INSERT的基础模板，比如INSERT INTO orders (user_id, user_name, customer_email, customer_mobile, amount, create_date, status_change_date) VALUES (?)。

3. 使用COPY命令实现极致性能 #

如果数据量极大（比如10万条以上），PostgreSQL的COPY命令是最快的方式，DataGrip快速插入大概率用的就是这个机制。R2DBC PostgreSQL驱动支持COPY操作，示例代码如下：

public Mono<Boolean> insertOrdersWithCopy(List<ClientOrder> orders) {
    if (orders.isEmpty()) {
        return Mono.just(true);
    }
    return databaseClient.inConnection(connection -> {
        // 创建COPY语句，指定目标表和列顺序
        CopyStatement copy = connection.createCopyStatement(
                "COPY orders (user_id, user_name, customer_email, customer_mobile, amount, create_date, status_change_date) FROM STDIN CSV");
        
        // 将数据转换成CSV格式写入
        return Flux.fromIterable(orders)
                .map(order -> String.join(",",
                        order.getUserId().toString(),
                        "\"" + order.getUserName().replace("\"", "\"\"") + "\"",
                        "\"" + order.getCustomerEmail().replace("\"", "\"\"") + "\"",
                        "\"" + order.getCustomerMobile().replace("\"", "\"\"") + "\"",
                        order.getAmount().toString(),
                        order.getCreateDate().toString(),
                        order.getStatusChangeDate().toString()
                ))
                .concatWith(Mono.just("\n")) // 写入结束符
                .flatMap(line -> Mono.from(copy.write(line.getBytes(StandardCharsets.UTF_8))))
                .then(Mono.from(copy.complete()))
                .map(result -> result.getRowsCopied() == orders.size());
    });
}

COPY命令直接把数据以流的形式写入数据库，避免了普通INSERT的SQL解析和参数绑定开销，性能提升非常明显。

4. 连接池配置优化 #

确保R2DBC连接池有足够的容量处理批量操作，在application.yml中调整：

spring:
  r2dbc:
    pool:
      max-size: 10
      initial-size: 5

足够的连接数可以避免等待连接的开销，提升批量操作的并行处理能力。

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