我尝试将原生的Neo4j Reactive Streams ReactiveSession 与reactor搭配，以完成 insert/update 和 findById 操作。

@Component
@RequiredArgsConstructor
@Slf4j
public class ProductRepository {
    private final Driver driver;

    Mono<Product> save(Product product) {
        String query = """
                MERGE (p:Product {id: $id})
                ON CREATE SET p.name=$name, p.price=$price
                ON MATCH SET p.name=$name, p.price=$price
                RETURN p.id as id, p.name as name, p.price as price
                """;
        Map<String, Object> parameters = Map.of("id", product.id() != null ? product.id() : UUID.randomUUID().toString(),
                "name", product.name(),
                "price", Values.value(product.price().toString())
        );

        SessionConfig sessionConfig = SessionConfig.builder()
                .withBookmarkManager(driver.executableQueryBookmarkManager())
                .build();

        // Open, execute, and automatically close a session for this specific call
        return Mono.usingWhen(
                Mono.fromSupplier(() -> driver.session(ReactiveSession.class, sessionConfig)),
                session -> Mono.from(session.executeWrite(tc -> Mono.from(tc.run(query, parameters))
                        .flatMapMany(ReactiveResult::records)
                        .single()
                        .map(result -> {
                            log.debug("saving product {}", result);
                            return new Product(
                                    result.get("id").asString(),
                                    result.get("name").asString(),
                                    new BigDecimal(result.get("price").asString())
                            );
                        })
                )),
                ReactiveSession::close // Closes the session on success/completion
        );
    }

    Mono<Product> findById(String id) {
        String query = """
                MATCH (p:Product)
                WHERE p.id=$id
                // Return the full properties map instead of property individual paths
                // RETURN p.id as id, p.name as name, p.price as price
                RETURN p { .id, .name, .price } as productData
                """;
        Map<String, Object> parameters = Map.of("id", id);
        SessionConfig sessionConfig = SessionConfig.builder()
                .withBookmarkManager(driver.executableQueryBookmarkManager())
                .build();

        // Open, execute, and automatically close a session for this specific call
        return Mono.usingWhen(
                Mono.fromSupplier(() -> driver.session(ReactiveSession.class, sessionConfig)),
                session -> Mono.from(session.executeRead(tc -> Mono.from(tc.run(query, parameters))
                        .flatMapMany(ReactiveResult::records)
                        .next()
                        .map(result -> {
                            // Extracting using standard .get("productData").get("id") syntax
                            var data = result.get("productData");
                            return new Product(
                                    data.get("id").asString(),
                                    data.get("name").asString(),
                                    new BigDecimal(data.get("price").asString())
                            );
                        })
                )),
                ReactiveSession::close // Closes the session on success/completion
        );
    }
}

并编写了一个测试来验证该功能。

@SneakyThrows
    @Test
    public void testProductRepository() {
        AtomicReference<String> idReference = new AtomicReference<>();

        // First, completely execute and complete the save transaction
        productRepository.save(new Product(null, "test", BigDecimal.ONE))
                .as(StepVerifier::create)
                .consumeNextWith(product -> {
                    assertThat(product).isNotNull();
                    assertThat(product.id()).isNotNull();
                    idReference.set(product.id());
                })
                .verifyComplete(); // The driver captures the write bookmark here

        var savedId = idReference.get();
        log.debug("savedId={}", savedId);
        assertThat(savedId).isNotNull();

        Thread.sleep(Duration.ofMillis(1_000));

        // The read session automatically uses that bookmark to guarantee casual consistency
        productRepository.findById(savedId)
                .delayElement(Duration.ofMillis(500))
                .as(StepVerifier::create)
                .consumeNextWith(p -> {
                    log.debug("found product by savedId: {}", p);
                    assertThat(p.name()).isEqualTo("test");
                })
                .verifyComplete(); // Suc
    }

从日志来看，保存没有问题，但在 findById consume With 行发生了异常，似乎没有找到数据。

对于 ProductRepository 的实现，我尝试了另一种方案：在bean实例化时初始化一个 ReactiveSession，也就是 @PostConstruct，并在整个 ProductRepository 组件中 共享 它。问题在于它会抛出关于事务的异常，测试中在对 findById 查询时，提示第一个TX仍然处于打开状态。

示例项目可以在这里找到：https://github.com/hantsy/spring-reactive-sample/tree/master/boot-neo4j

解决方案

从Neo4j Java SDK的 GitHub问题中找到了原因，详情请看 这里。

在 session.executeWrite 块中，事务通过在同一个 Publisher 中再发送一个回调项来提交。并且使用 Mono.from 从通用的 Publisher 构建一个 Mono 时，会丢弃同一发布者中的后续项。Mono.fromDirect 略有不同，它在使用第一条已发出的项来构建 Mono 时，并不会阻止原始发布者继续发出项。

改为 Mono.fromDirect 或 Flux.from 解决了这个问题。