Как контролировать поток транзакционного потока с Spring Data R2DBC?

Question

Поддержка транзакционных потоков, похоже, была недавно реализована, но из-за ее новизны примеров кода не так много.

Может ли кто-нибудь показать пример транзакционного потока, который выполняет серию вставок в базу данных, а затем возвращает определенное значение в случае успеха, но с промежуточной контрольной точкой между вставками, которая проверяет некоторое условие и может откатить транзакцию и вернуть разные значения в зависимости от контрольной точки результат?

Answer 1

Реактивные транзакции следуют той же схеме, что и императивные:

1. Транзакция запускается перед выполнением любых команд пользовательского пространства
2. Запустите команды пространства пользователя
3. Фиксация (или откат)

Здесь следует отметить несколько аспектов: Соединение всегда связано с материализацией реактивной последовательности. Что мы знаем из Threadсвязанное соединение, которое связано с выполнением в императивном программировании, означает материализацию в реактивном программировании.

Таким образом, каждое (одновременное) выполнение получает назначенное соединение.

Spring Data R2DBC не поддерживает точки сохранения. Взгляните на следующий пример кода, который иллюстрирует решение о коммите или откате:

DatabaseClient databaseClient = DatabaseClient.create(connectionFactory);

TransactionalOperator transactionalOperator = TransactionalOperator
        .create(new R2dbcTransactionManager(connectionFactory));

transactionalOperator.execute(tx -> {

    Mono<Void> insert = databaseClient.execute("INSERT INTO legoset VALUES(…)")
            .then();

    Mono<Long> select = databaseClient.execute("SELECT COUNT(*) FROM legoset")
            .as(Long.class)
            .fetch()
            .first();

    return insert.then(select.handle((count, sink) -> {

        if(count > 10) {
            tx.setRollbackOnly();
        }

    }));
}).as(StepVerifier::create).verifyComplete();

Примечательные аспекты здесь:

1. Мы используем TransactionalOperator вместо @Transactional,
2. Код в .handle() звонки setRollbackOnly() откатить транзакцию.

С помощью @Transactionalобычно вы используете исключения для сигнализации об условиях отката.