如何在 r2dbc 中批量执行多个插入？

Posted 2023-02-25

【中文标题】如何在 r2dbc 中批量执行多个插入？

【英文标题】：How to execute multiple inserts in batch in r2dbc?

【发布时间】：2020-10-12 07:04:47

【问题描述】：

我需要在一个表中批量插入多行。 在 DatabaseClient 中，我找到了 insert() 语句和 using(Publisher objectToInsert) 方法，该方法具有多个对象作为参数。但它会在一批中插入它们吗？ 另一种可能的解决方案是connection.createBatch()，但它有一个缺点：我不能在那里传递我的实体对象，也不能从实体生成sql查询。

那么，是否可以在 r2dbc 中创建批量插入？

【参考方案1】：

有两个问题：

DatabaseClient.insert() 是否会在一批中插入它们？

不是一个批次。

是否可以在 r2dbc 中创建批量插入？ （Connection.createBatch()除外）

不，使用Connection.createBatch() 只是现在创建Batch 的一种方法。

另见问题：spring-data-r2dbc#259

【讨论】：

如果INSERT命令相同，Connection.createStatement优于Connection.createBatch。见r2dbc.io/spec/0.8.5.RELEASE/spec/html/#statements.batching

【参考方案2】：

目前还没有直接的支持，但是我发现可以使用Connection来简单地克服这个障碍，看看这个问题，spring-data-r2dbc#259

statement 有一个add 可以重复绑定参数。

我的解决方案的完整代码可以在here找到。

        return this.databaseClient.inConnectionMany(connection -> 

            var statement = connection.createStatement("INSERT INTO  posts (title, content) VALUES ($1, $2)")
                    .returnGeneratedValues("id");

            for (var p : data) 
                statement.bind(0, p.getTitle()).bind(1, p.getContent()).add();
            

            return Flux.from(statement.execute()).flatMap(result -> result.map((row, rowMetadata) -> row.get("id", UUID.class)));
        );

对此方法的测试。

@Test
    public void testSaveAll() 

        var data = Post.builder().title("test").content("content").build();
        var data1 = Post.builder().title("test1").content("content1").build();

        var result = posts.saveAll(List.of(data, data1)).log("[Generated result]")
                .doOnNext(id->log.info("generated id: ", id));

        assertThat(result).isNotNull();
        result.as(StepVerifier::create)
                .expectNextCount(2)
                .verifyComplete();
    

生成的 id 会在控制台中按预期打印。

...
2020-10-08 11:29:19,662 INFO [reactor-tcp-nio-2] reactor.util.Loggers$Slf4JLogger:274 onNext(a3105647-a4bc-4986-9ad4-1e6de901449f)
2020-10-08 11:29:19,664 INFO [reactor-tcp-nio-2] com.example.demo.PostRepositoryTest:31 generated id: a3105647-a4bc-4986-9ad4-1e6de901449f
//.....
2020-10-08 11:29:19,671 INFO [reactor-tcp-nio-2] reactor.util.Loggers$Slf4JLogger:274 onNext(a611d766-f983-4c8e-9dc9-fc78775911e5)
2020-10-08 11:29:19,671 INFO [reactor-tcp-nio-2] com.example.demo.PostRepositoryTest:31 generated id: a611d766-f983-4c8e-9dc9-fc78775911e5
//......

Process finished with exit code 0

【讨论】：

此解决方案是否使用批处理？

请注意，某些驱动程序（例如 mariadb）需要问号“？，？”而不是 "$1, $2" 以使绑定起作用。

@Readren 每个数据库在 SQL 中都有自己的参数占位符，例如。 mssql 使用@ 前缀，pg 使用$ 等

重要提示：为确保插入确实发生，必须在 each 上调用 getRowsUpdated() 或 map(...)返回Result 并将其结果包含在反应链中。仅仅做 Flux.from(statement.execute()).then() 这样的事情是不够的。