Event-driven архитектура на Kotlin: от корутин до реактивных стримов

Event-driven архитектура на Kotlin: почему это не просто модный тренд

Представьте, что ваше приложение — это идеально отлаженный механизм. Каждая шестерёнка знает, когда ей крутиться, каждый поршень — когда двигаться. Но что, если однажды потребуется добавить новую деталь, не останавливая весь станок? В традиционной синхронной архитектуре это часто превращается в кошмар: переписывание целых модулей, тестирование бесконечных цепочек вызовов и бессонные ночи.

Здесь на сцену выходит event-driven архитектура на Kotlin — парадигма, где компоненты общаются через события, а не через прямые вызовы. Kotlin, с его корутинами и реактивными стримами, даёт для этого просто идеальные инструменты. В этой статье мы не просто поговорим о теории — мы напишем реальный код, разберём Kotlin Flow и Channels, и я покажу, как построить гибкую, масштабируемую систему, которая не рассыплется при первой же нагрузке.

Что такое событийно-ориентированная архитектура простыми словами

В классическом подходе (представьте себе микросервисы, которые звонят друг другу по REST) каждый запрос — это жёсткая связь. Сервис А вызывает сервис Б, ждёт ответа, и если Б упал — всё ломается. В event-driven архитектуре всё иначе: компоненты публикуют события (например, "Заказ создан") и подписываются на них. Никто никого не ждёт, никто никому не звонит напрямую.

Прелесть Kotlin в том, что он позволяет реализовать эту модель нативно, без подключения громоздких фреймворков. Корутины дают лёгковесные потоки, а Channel и Flow — готовые шины для событий.

Давайте посмотрим на простейший пример — свой EventBus на Kotlin:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

class SimpleEventBus {
    private val events = Channel<Any>(Channel.UNLIMITED)

    suspend fun publish(event: Any) {
        events.send(event)
    }

    suspend fun subscribe(handler: suspend (Any) -> Unit) {
        for (event in events) {
            handler(event)
        }
    }
}

// Использование
fun main() = runBlocking {
    val bus = SimpleEventBus()
    
    launch {
        bus.subscribe { event ->
            println("Получено событие: $event")
        }
    }
    
    delay(100)
    bus.publish("Hello, Event-Driven!")
    delay(100)
}

Этот код — база. Мы создаём канал, который работает как асинхронная очередь. Издатель кладёт событие, подписчик его забирает. Всё просто, но уже здесь видна сила event-driven архитектуры на Kotlin: подписчик и издатель ничего не знают друг о друге.

Kotlin Flow: реактивные стримы для событий

Channel — это здорово, но для реальных проектов часто нужно больше: холодные и горячие стримы, трансформации, обработка ошибок. Здесь в игру вступает Kotlin Flow — реактивная библиотека, встроенная прямо в язык.

Flow идеально подходит для асинхронного программирования на Kotlin, когда события генерируются постоянно. Например, обновления курса валют, ленты новостей или состояния UI.

Холодные vs горячие стримы

Важно понимать разницу:

• Холодный стрим (Cold Stream) — начинает работать только когда на него подписались. Каждый подписчик получает свой поток данных.
• Горячий стрим (Hot Stream) — работает постоянно, независимо от подписчиков. Поздние подписчики пропускают старые события.

Для event-driven архитектуры обычно нужны горячие стримы. В Kotlin их реализуют через SharedFlow или StateFlow.

Пример — система уведомлений:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

data class Notification(val userId: String, val message: String)

class NotificationService {
    private val _notifications = MutableSharedFlow<Notification>()
    val notifications: SharedFlow<Notification> = _notifications.asSharedFlow()

    suspend fun sendNotification(userId: String, message: String) {
        _notifications.emit(Notification(userId, message))
    }
}

fun main() = runBlocking {
    val service = NotificationService()
    
    // Подписчик 1
    launch {
        service.notifications.collect { 
            println("[Подписчик 1] ${it.userId}: ${it.message}")
        }
    }
    
    // Подписчик 2 (пропустит первые события)
    delay(50)
    launch {
        service.notifications.collect {
            println("[Подписчик 2] ${it.userId}: ${it.message}")
        }
    }
    
    service.sendNotification("user1", "Привет!")
    delay(30)
    service.sendNotification("user2", "Как дела?")
    delay(100)
}

Обратите внимание: SharedFlow — это горячий стрим. Если подписчик подключился позже, он не получит предыдущие события. Это идеально для реактивных стримов Kotlin в реальном времени.

Channels: надёжная передача событий между корутинами

Если Flow — это стрим для "широковещательных" событий, то Channels — это точка-точка. Каждое событие получает ровно один подписчик. Это как очередь сообщений внутри приложения.

Пример: пул воркеров, обрабатывающих задачи:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

data class Task(val id: Int, val payload: String)

suspend fun worker(id: Int, channel: Channel<Task>) {
    for (task in channel) {
        println("Воркер $id обрабатывает задачу ${task.id}: ${task.payload}")
        delay(100) // имитация работы
    }
}

fun main() = runBlocking {
    val taskChannel = Channel<Task>(Channel.UNLIMITED)
    
    // Запускаем 3 воркера
    repeat(3) { id ->
        launch { worker(id, taskChannel) }
    }
    
    // Отправляем задачи
    repeat(10) { i ->
        taskChannel.send(Task(i, "Данные #$i"))
    }
    
    delay(500)
    taskChannel.close()
}

Здесь Channel работает как балансировщик: задачи распределяются между воркерами. Это классический пример событийно-ориентированной архитектуры в действии — продюсер не знает, кто именно обработает задачу.

Продвинутые паттерны: Event Sourcing и CQRS на Kotlin

Когда мы говорим про event-driven архитектуру на Kotlin в enterprise-проектах, нельзя обойти стороной Event Sourcing. Вместо хранения текущего состояния, мы храним всю историю событий. А чтобы читать эти события эффективно, используем CQRS (Command Query Responsibility Segregation).

Kotlin с его корутинами делает реализацию таких паттернов удивительно элегантной. Вот как может выглядеть простое Event Store:

import kotlinx.coroutines.flow.*

data class Event(val aggregateId: String, val type: String, val data: String)

class EventStore {
    private val