Как обрабатывать транзакции в ParallelFlux? ⇐ JAVA

[Anonymous]

У нас есть очень перерабатывающий конвейер для тяжелых флюсов. Чтобы ускорить его, мы используем параллельный поток. Но теперь проблема в том, что все ожидает подключения к базе данных и все еще медленно. Весь конвейер в настоящее время находится в рамках одной большой транзакции. Как обрабатывать параллельныйFlux с транзакциями, чтобы вставки в БД не слишком сильно блокировали друг друга? База данных — Postgresql, мы используем r2dbc. Или еще лучше, если разные параллельные ветви будут использовать разные соединения.
Упрощенный пример

Код: Выделить всё

@Transaction
public void someFunc(){
doSomething()
.buffer()
.flatMapIterable(x -> x)
.parallel(16)
.concatMap(id -> doSomethingInParralel(id))
.sequential()
.collectList();
}

public Object doSomethingInParalel(int id){
// run some calculation heavy code here
repository.insertData(some calculated data);
}

Желаемым результатом будет то, что someFunc находится в транзакции. doSomethingInParralel следует запускать параллельно друг другу, например 10 параллельно или 100 параллельно, и каждый из них должен иметь свою собственную транзакцию, чтобы они не блокировали друг друга при попытке чтения или записи базы данных.
Я пытаюсь добавить Propagation.REQUIRE_NEW в doSomethingInParralel, но проблема в том, что первая создаваемая новая транзакция приостанавливает исходную транзакцию. с которым работает someFunc. И поскольку оригинал приостановлен, я не могу создавать больше параллельных транзакций. использование Propagation.NESTED не является вариантом, поскольку оно использует то же соединение, что и его родительская транзакция.
Как мне создать параллельные транзакции с отдельными подключениями к базе данных?< /п>

Подробнее здесь: https://stackoverflow.com/questions/792 ... rallelflux