Сегодня коротко поговорим про Spring Event Bus и разберем одну задачку.

Недавно столкнулся с ситуацией - есть некий общий функционал, который вызывается из совершенно разных роутов. Сами роуты обернуты в транзакцию. Сообщение (ивент) в брокер об успешном выполнении этого общего функционала нужно отправлять прямо внутри этого небольшого метода. Но чтобы всё отработало безопасно, ивент должен улетать только после коммита транзакции.

Конечно, первым на ум приходит запихивание отправки ивента в конец каждого верхнего роута. А хотелось бы просто опубликовать событие локально и быть уверенным, что оно улетит после успешного коммита внешней транзакции, кем бы она ни открывалась. И Spring event bus вместе с аннотацией @TransactionalEventListener дают возможное решение данной ситуации.

Шина событий в Spring - это такой встроенный publisher/subscriber внутри одного приложения: какой-то компонент системы может отправить событие в шину, а настроенные слушатели обработают его заданным образом.

Создать событие в Kotlin можно проще простого, обычным data class:

data class ActionCompletedEvent(
    val entityId: Long,
    val payload: String
)

В сервисе мы публикуем событие ровно там, где оно должно происходить по логике, и забываем про него. Публикуется оно, кстати, через ApplicationEventPublisher:

@Service
class SomeService(
    private val someRepository: SomeRepository,
    private val eventPublisher: ApplicationEventPublisher
) {

    // Метод вызывается из разных транзакционных роутов
    fun performAction(entity: Entity, payload: String) {
        someRepository.save(...)
        // Публикуем событие локально. 
        // Что будет дальше - нас не волнует.
        eventPublisher.publishEvent(ActionCompletedEvent(entity.id, payload))
    }
}

Если мы в слушателе повесим обычный @EventListener, он сработает прямо в момент вызова publishEvent. То есть проблема с откатом транзакции останется -  ивент улетит до того, как внешняя транзакция подтвердит сохранение. Но у Spring есть @TransactionalEventListener:

@Component
class ActionListener(
    private val kafka: kafkaClient
) {

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    fun handleActionCompleted(event: ActionCompletedEvent) {
        // Этот код выполнится ТОЛЬКО после успешного коммита в БД.
        // Spring сам поймет, к какой внешней транзакции мы привязаны.
        kafka.sendEvent(event.entityId, event.payload)
    }
}

Видите фазу AFTER_COMMIT? Spring сам отслеживает контекст транзакции. Мы можем легально вызывать publishEvent в любом месте метода, сохраняя правильную семантику кода. А Spring гарантирует, что слушатели отработают только после успешного коммита самой верхней транзакции в стеке вызовов. Если что-то упадет и транзакция откатится, событие просто игнорируется.

Единственный нюанс при использовании - это то, что по умолчанию @TransactionalEventListener работает синхронно, в том же потоке, что и основной запрос. Если обработка ивента займет пару секунд, юзер будет сидеть и ждать ответа от сервера, хотя в базу данные уже записаны. Для решения данного казуса можно воспользоваться аннотацией @Async над слушателем.

Итого

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