Программное управление

TransactionTemplate - транзакция в коде, без аннотации. Аннотации удобнее, программное управление точнее.

@Service
public class OrderService {

    private final TransactionTemplate txTemplate;

    public OrderService(PlatformTransactionManager txManager) {
        this.txTemplate = new TransactionTemplate(txManager);
        this.txTemplate.setIsolationLevel(TransactionDefinition.ISOLATION_REPEATABLE_READ);
        this.txTemplate.setTimeout(10);
    }

    public Order createOrder(OrderRequest req) {
        return txTemplate.execute(status -> {
            try {
                Order order = orderRepo.save(new Order(req));
                inventoryService.reserve(order);
                return order;
            } catch (InsufficientInventoryException e) {
                status.setRollbackOnly();   // откат без проброса исключения из транзакции
                throw e;
            }
        });
    }
}

Когда лучше аннотаций:

• сложная ветвящаяся логика, параметры транзакции выбираются динамически
• нужна узкая граница вокруг маленькой критической секции
• нужно обойти ограничения прокси (self-invocation, утилитные классы)
• нужно явно показать транзакционный поток в коде
• тесты

В реактивном стеке то же самое делает TransactionalOperator.

---

Действия после коммита

TransactionSynchronization - хуки жизненного цикла текущей транзакции.

@Transactional
public void placeOrder(Order order) {
    orderRepo.save(order);

    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
        @Override
        public void afterCommit() {
            eventPublisher.publish(new OrderCreatedEvent(order.getId()));   // данные уже в БД
        }
    });
}

@TransactionalEventListener - то же декларативно (Spring 4.2+). По умолчанию слушатель срабатывает в фазе AFTER_COMMIT.

@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
public void onOrderCreated(OrderCreatedEvent event) {
    notificationService.notify(event);
}

Полезно для отправки доменных событий и любой логики «реагируй на уже закоммиченное состояние». Это не распределённая транзакция: если приложение упадёт между коммитом и отправкой, событие потеряется. Надёжная доставка - через Outbox.

---

TransactionSynchronizationRegistry (JTA) - тот же приём в JTA-транзакции Jakarta EE. Пример из лабы: файл кладётся в MinIO перед коммитом БД, а при откате удаляется (см. Best-effort 2PC).

ChainedTransactionManager (deprecated, но встречается) - коммитит несколько менеджеров по очереди. Это не атомарно: если второй коммит упал, первый уже закоммичен. Best effort, а не распределённая транзакция.

@Bean
public PlatformTransactionManager chainedTxManager(DataSourceTransactionManager ds1TxManager,
                                                   DataSourceTransactionManager ds2TxManager) {
    return new ChainedTransactionManager(ds1TxManager, ds2TxManager);
}

---

Маршрутизация DataSource

AbstractRoutingDataSource - выбирает DataSource на каждый запрос. Типично: запись на primary, чтение на реплику по флагу readOnly.

public class ReadWriteRoutingDataSource extends AbstractRoutingDataSource {
    @Override
    protected Object determineCurrentLookupKey() {
        return TransactionSynchronizationManager.isCurrentTransactionReadOnly() ? "replica" : "primary";
    }
}

@Bean
public DataSource routingDataSource(DataSource primaryDataSource, DataSource replicaDataSource) {
    ReadWriteRoutingDataSource ds = new ReadWriteRoutingDataSource();
    ds.setTargetDataSources(Map.of(
            "primary", primaryDataSource,
            "replica", replicaDataSource
    ));
    ds.setDefaultTargetDataSource(primaryDataSource);
    return ds;
}

Note

Чтобы ключ вычислялся уже с известным флагом readOnly, routing-DataSource оборачивают в LazyConnectionDataSourceProxy: соединение берётся при первом запросе, а не в начале транзакции.

---

Транзакции и потоки

Одну imperative-транзакцию нельзя разделить между потоками: контекст в ThreadLocal текущего потока. Новый поток - новый контекст.

@Transactional
public void processInParallel(List<Item> items) {
    items.parallelStream().forEach(item -> processItem(item));   // НЕПРАВИЛЬНО: потоки без транзакции вызывающего
    CompletableFuture.supplyAsync(() -> heavyComputation(item));  // НЕПРАВИЛЬНО: транзакция не наследуется
}

Правильно: у каждого параллельного обработчика своя транзакционная граница. Согласованность между задачами обеспечивают не общей транзакцией, а координацией: барьер, saga, конечный автомат, outbox, идемпотентное слияние.

Подход 1 - каждый поток вызывает транзакционный метод другого бина:

@Service
public class BatchService {
    private final ItemProcessor itemProcessor;   // отдельный бин - свой прокси - своя транзакция

    public void processBatch(List<Item> items) {
        List<CompletableFuture<Void>> futures = items.stream()
                .map(item -> CompletableFuture.runAsync(() -> itemProcessor.processItem(item), taskExecutor))
                .toList();
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
    }
}

@Service
public class ItemProcessor {
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void processItem(Item item) {
        // каждый вызов в своей транзакции
    }
}

Подход 2 - TransactionTemplate в рабочем потоке:

public void processBatch(List<Item> items) {
    ExecutorService executor = Executors.newFixedThreadPool(4);
    List<Future<?>> futures = items.stream()
            .map(item -> executor.submit(() ->
                    txTemplate.execute(status -> {   // транзакция создаётся в рабочем потоке
                        processItem(item);
                        return null;
                    })))
            .toList();

    for (Future<?> f : futures) {
        try {
            f.get();
        } catch (ExecutionException e) { /* обработка ошибки задачи */ }
    }
}

Подход 3 - @Async + @Transactional:

@Service
public class AsyncProcessor {
    @Async("taskExecutor")   // новый поток
    @Transactional           // транзакция в этом потоке
    public CompletableFuture<Result> processAsync(Item item) {
        return CompletableFuture.completedFuture(doWork(item));
    }
}

Работает, только если оба прокси выстроены правильно. Надёжнее разнести @Async и @Transactional по разным бинам.

---

Virtual threads (Java 21+):

@Bean
public TaskExecutor taskExecutor() {
    return new TaskExecutorAdapter(Executors.newVirtualThreadPerTaskExecutor());
}

• ThreadLocal у виртуального потока свой, транзакционный контекст привязывается так же, как к обычному
• до Java 24 блок synchronized внутри JDBC-драйвера «пинил» carrier-поток. Драйвер PostgreSQL заменил synchronized на ReentrantLock, у других драйверов проверяй версию. С Java 24 (JEP 491) synchronized больше не пинит

Чего не делать:

• ждать, что @Async «унаследует» транзакцию
• держать долгую транзакцию и распараллеливать внутри неё операции с БД
• ходить в сеть, ждать брокер или долго считать внутри транзакции

Чем дольше живёт транзакция, тем больше ожидание блокировок, дедлоков, нагрузка на пул соединений, шанс отката и хвостовые задержки. Особенно с REQUIRES_NEW, которому нужно второе соединение.

Итого: транзакция как можно короче, критическая секция как можно уже.