使用带有事务的反应性(r2dbc)批处理

Question

嗨，我有一个非常重要的问题。我正在尝试使用reactive r2dbc创建一个批处理，并使用transactional来注释该方法。但是看起来，如果我同时使用事务性代码和批处理代码，代码就会挂起，不起作用。下面是代码

    @Transactional /* **Transaction** */
    @GetMapping("/batchFetchData")
    public Flux<Object> batchFetch() {
        long startTime = System.currentTimeMillis();
        Mono.from(databaseConfiguration.connectionFactory().create())
                .flatMapMany(connection -> Flux.from(connection
                        .createBatch() /* **Creating batch***/
                        .add("SELECT * FROM xtable where xId = 232323")
                        .add("SELECT * FROM ytable where yId = 454545")
                        .add("SELECT * FROM ztable where zId = 676767")
                        //.execute()));  /* **Execution batch***/
                        .execute())).as(StepVerifier::create)
                .expectNextCount(3) /* **Expect count batch***/
                .verifyComplete();  /* **Verify batch***/

        LOGGER.info("Time taken to batchFetch "+(System.currentTimeMillis() - startTime));
    return null;
    }

Answer 1

你在破坏反应链。

在反应式编程中，在订阅之前不会发生任何事情。

这是什么意思，我可以举一个小例子来说明。

// If running this, nothing happens
Mono.just("Foobar");

Mono.just("Foobar").subscribe(s -> System.out.println(s));

Foobar

public void getString() {
    Mono.just("Foobar");
}

// Nothing happens, you have declared something 
// but it will never get run, no one is subscribing
getString();

public Mono<String> getString() {
    // This could be saving to a database or anything, this will now get run
    return Mono.just("Now this code will get run");
}

// The above got run, we can prove it by printing
getString().subscribe(s -> System.out.println(s));

如果没有人订阅，就不会建立链。

那么订户是谁？它通常是价值的最终消费者。例如，启动调用的网页或移动应用程序，但也可以是您的Spring Boot服务，如果它是启动调用的服务（例如在cron作业中）。

因此，让我们看看您的代码：

@Transactional /* **Transaction** */
@GetMapping("/batchFetchData")
public Flux<Object> batchFetch() {
    long startTime = System.currentTimeMillis();

    // Here you declare a Mono but ignoring the return type so breaking the reactive chain
    Mono.from(databaseConfiguration.connectionFactory().create()) 
            .flatMapMany(connection -> Flux.from(connection
                    .createBatch() /* **Creating batch***/
                    .add("SELECT * FROM xtable where xId = 232323")
                    .add("SELECT * FROM ytable where yId = 454545")
                    .add("SELECT * FROM ztable where zId = 676767")
                    //.execute()));  /* **Execution batch***/
                    .execute())).as(StepVerifier::create)
            .expectNextCount(3) /* **Expect count batch***/
            .verifyComplete();  /* **Verify batch***/
            // Here at the end you have no subscriber

    LOGGER.info("Time taken to batchFetch "+(System.currentTimeMillis() - startTime));

    // Null is not allowed in reactive chains
    return null;
}

@Transactional
@GetMapping("/batchFetchData")
public Mono<Void> batchFetch() {
    long startTime = System.currentTimeMillis();

    // we return here so that the calling client 
    // can subscribe and start the chain
    return Mono.from(databaseConfiguration.connectionFactory().create()) 
            .flatMapMany(connection -> Flux.from(connection
                    .createBatch()
                    .add("SELECT * FROM xtable where xId = 232323")
                    .add("SELECT * FROM ytable where yId = 454545")
                    .add("SELECT * FROM ztable where zId = 676767")
                    .execute()))
                    .then(); 
                    // then() statement throws away whatever the return 
                    // value is and just signals to the calling client 
                    // when everything is done.       
}

这就是mono#then语句的作用。您可以看到，当链中的每个部分完成时，它将发出完成的信号，然后将值从一个部分传递到下一个部分，然后再次发出信号，并传递值等。当我们到达then语句时，它只发出完成的信号，而不返回任何东西（或者实际上它返回一个mono ，因为在反应链中不允许null）。您必须始终返回，以便每一步都能传递其完整的信号。

另外，我删除了代码中使用的StepVerifier，因为它通常用于验证单元测试中的步骤，而不是在生产代码中使用。你可以在这里阅读更多关于它的信息。

如果你想学习反应编程，我建议你这样做，因为它很神奇，我喜欢它，我强烈建议你阅读优秀的反应编程文档介绍，在那里他们将解释什么都不发生的概念，直到你订阅等。