java - 将 Stream 包装到 Flux 时的事务处理

Question

在手动包装 Stream 时，我真的很难理解背后发生的事情。作为来自 spring data jpa 的查询结果接收到 Flux .

考虑以下:

实体:

@NoArgsConstructor
@AllArgsConstructor
@Data
@Entity
public class TestEntity {
    @Id
    private Integer a;
    private Integer b;
}

存储库:

public interface TestEntityRepository extends JpaRepository<TestEntity, Integer> {
    Stream<TestEntity> findByBBetween(int b1, int b2);
}

简单的测试代码:

@Test
@SneakyThrows
@Transactional
public void dbStreamToFluxTest() {
    testEntityRepository.save(new TestEntity(2, 6));
    testEntityRepository.save(new TestEntity(3, 8));
    testEntityRepository.save(new TestEntity(4, 10));

    testEntityFlux(testEntityStream()).subscribe(System.out::println);
    testEntityFlux().subscribe(System.out::println);
    Thread.sleep(200);
}

private Flux<TestEntity> testEntityFlux() {
    return fromStream(this::testEntityStream);
}

private Flux<TestEntity> testEntityFlux(Stream<TestEntity> testEntityStream) {
    return fromStream(() -> testEntityStream);
}

private Stream<TestEntity> testEntityStream() {
    return testEntityRepository.findByBBetween(1, 9);
}

static <T> Flux<T> fromStream(final Supplier<Stream<? extends T>> streamSupplier) {
    return Flux
            .defer(() -> Flux.fromStream(streamSupplier))
            .subscribeOn(Schedulers.elastic());
}

问题:

这是做我所做的正确方法吗，尤其是关于静态 fromStream方法？

调用testEntityFlux(testEntityStream())做我所期望的，出于我真的不明白的原因，对 testEntityFlux() 的调用遇到错误:

reactor.core.Exceptions$ErrorCallbackNotImplemented: org.springframework.dao.InvalidDataAccessApiUsageException: You're trying to execute a streaming query method without a surrounding transaction that keeps the connection open so that the Stream can actually be consumed. Make sure the code consuming the stream uses @Transactional or any other way of declaring a (read-only) transaction. Caused by: org.springframework.dao.InvalidDataAccessApiUsageException: You're trying to execute a streaming query method without a surrounding transaction that keeps the connection open so that the Stream can actually be consumed. Make sure the code consuming the stream uses @Transactional or any other way of declaring a (read-only) transaction.

...当我忘记@Transactional 时通常会发生什么，而我没有。

编辑

注意:代码的灵感来自:https://github.com/chang-chao/spring-webflux-reactive-jdbc-sample/blob/master/src/main/java/me/changchao/spring/springwebfluxasyncjdbcsample/service/CityServiceImpl.java这反过来又受到 https://spring.io/blog/2016/07/20/notes-on-reactive-programming-part-iii-a-simple-http-server-application 的启发.
但是，Mono版本具有相同的“问题”。

编辑 2

使用可选的示例，注意 testEntityMono()替换 testEntityOptional()与 testEntityOptionalManual()导致工作代码。因此，这一切似乎都与 jpa 如何获取数据直接相关:

@SneakyThrows
@Transactional
public void dbOptionalToMonoTest() {
    testEntityRepository.save(new TestEntity(2, 6));
    testEntityRepository.save(new TestEntity(3, 8));
    testEntityRepository.save(new TestEntity(4, 10));

    testEntityMono(testEntityOptional()).subscribe(System.out::println);
    testEntityMono().subscribe(System.out::println);

    Thread.sleep(1200);
}

private Mono<TestEntity> testEntityMono() {
    return fromSingle(() -> testEntityOptional().get());
}

private Mono<TestEntity> testEntityMono(Optional<TestEntity> testEntity) {
    return fromSingle(() -> testEntity.get());
}

private Optional<TestEntity> testEntityOptional() {
    return testEntityRepository.findById(4);
}

@SneakyThrows
private Optional<TestEntity> testEntityOptionalManual() {
    Thread.sleep(1000);
    return Optional.of(new TestEntity(20, 20));
}

static <T> Mono<T> fromSingle(final Supplier<T> tSupplier) {
    return Mono
            .defer(() -> Mono.fromSupplier(tSupplier))
            .subscribeOn(Schedulers.elastic());
}

Answer 1

TL;博士:

它归结为命令式和响应式(Reactive)编程假设与 Thread 之间的差异。亲和力。

细节

我们首先需要了解事务管理会发生什么，以了解为什么您的安排以失败告终。

使用 @Transactional方法为方法中的所有代码创建一个事务范围。返回标量值的事务方法，Stream ，类似集合的类型，或 void (基本上是非 react 性类型)被认为是命令式事务方法。

在命令式编程中，流程坚持其载体 Thread .代码预计保持不变Thread而不是切换线程。因此，事务管理将事务状态和资源与载体 Thread 相关联。在 ThreadLocal贮存。一旦事务方法中的代码切换线程(例如启动新的 Thread 或使用 Thread 池)，将在不同的 Thread 上执行的工作单元离开事务范围并可能在自己的事务中运行。在最坏的情况下，事务在外部 Thread 上保持打开状态。因为没有事务管理器监控事务工作单元的进入/退出。
@Transactional返回响应式类型的方法(例如 Mono 或 Flux )受响应式事务管理的约束。响应式(Reactive)事务管理不同于命令式事务管理，因为事务状态附加到 Subscription ，特别是订阅者 Context .上下文仅适用于 react 类型，不适用于标量类型，因为无法将数据附加到 void或 String .

看代码:

@Test
@Transactional
public void dbStreamToFluxTest() {
    // …
}

我们看到这个方法是 @Transactional测试方法。这里我们有两点需要考虑:

该方法返回void因此它受制于将事务状态与 ThreadLocal 关联的命令式事务管理.

@Test 没有反应式事务支持方法，因为通常是 Publisher预计将从该方法返回，并且这样做，将无法断言流的结果。

@Test
@Transactional
public Publisher<Object> thisDoesNotWork() {
    return myRepository.findAll(); // Where did my assertions go?
}

让我们仔细看看fromStream(…)方法:

static <T> Flux<T> fromStream(final Supplier<Stream<? extends T>> streamSupplier) {
    return Flux
        .defer(() -> Flux.fromStream(streamSupplier))
        .subscribeOn(Schedulers.elastic());
}

该代码接受 Supplier返回 Stream .接下来，订阅(subscribe(…)，request(…))信号被指示在弹性 Scheduler 上发生。这有效地打开了Thread Stream被创建和消费。因此，subscribeOn导致 Stream创建(调用 findByBBetween(…) )在不同的 Thread 上发生比您的运营商 Thread .

删除 subscribeOn(…)将解决您的问题。

还有一点可以解释为什么要避免在 JPA 中使用响应式类型。响应式(Reactive)编程没有强大的Thread亲和力。 Thread随时可能发生切换。取决于您如何使用生成的 Flux以及如何设计实体，当实体跨线程传递时，您可能会遇到可见性问题。理想情况下，响应式(Reactive)上下文中的数据保持不变。这种方法并不总是符合 JPA 规则。

另一方面是延迟加载。通过使用来自运营商以外线程的 JPA 实体 Thread ，实体可能无法将其上下文关联回 JPA 事务。您可以轻松遇到LazyInitializationException不知道为什么这是 Thread切换对您来说可能是不透明的。

建议是:不要将 react 类型与 JPA 或任何其他事务性资源一起使用。继续使用 Java 8 Stream反而。