Ktor: Server-Sent Events и WebSockets — реальное использование в продакшене

Задача была классическая: дашборд для операторов, который показывает новые заказы в реальном времени. Первая мысль — WebSockets, всё же это «реальное время». Но потом я посмотрел на требования: сервер отправляет данные, клиент только получает. Никакого двустороннего общения. Зачем стрелять из пушки по воробьям? SSE (Server-Sent Events) подошёл идеально.

А через месяц появилась задача чата между оператором и клиентом. Тут уже без WebSockets не обошлось. Так я реализовал оба подхода в одном Ktor-приложении и теперь могу сравнить на реальном опыте.

SSE в Ktor — проще, чем кажется

Ktor не имеет встроенной SSE-поддержки «из коробки» в виде отдельного плагина, но реализация на базе ручного OutgoingContent — дело 30 минут. Я использую подход с ручной отправкой через текстовый стрим:

import io.ktor.server.application.*
import io.ktor.server.response.*
import io.ktor.server.routing.*
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.serialization.Serializable
import kotlinx.serialization.json.Json

@Serializable
data class OrderNotification(
    val orderId: Long,
    val customerName: String,
    val total: Double,
    val type: String = "new_order"
)

class SseBroadcaster {
    private val channels = mutableListOf>()
    private val mutex = Mutex()

    suspend fun subscribe(channel: SendChannel) {
        mutex.withLock { channels.add(channel) }
    }

    suspend fun unsubscribe(channel: SendChannel) {
        mutex.withLock { channels.remove(channel) }
    }

    suspend fun broadcast(event: String) {
        mutex.withLock {
            channels.removeAll { channel ->
                val success = channel.trySend(event).isSuccess
                !success
            }
        }
    }
}

val sseBroadcaster = SseBroadcaster()

И маршрут для подключения клиента:

routing {
    get("/api/notifications/stream") {
        val channel = Channel(Channel.BUFFERED)
        sseBroadcaster.subscribe(channel)

        try {
            call.response.cacheControl("no-cache")
            call.response.header("Connection", "keep-alive")
            call.response.header("Content-Type", "text/event-stream")

            call.respondTextWriter(contentType = ContentType.Text.EventStream) {
                // Отправляем начальное событие подтверждения
                write("event: connected\ndata: {\"status\":\"ok\"}\n\n")
                flush()

                for (event in channel) {
                    write("event: order\ndata: $event\n\n")
                    flush()
                }
            }
        } finally {
            sseBroadcaster.unsubscribe(channel)
            channel.close()
        }
    }
}

Ключевой момент: SSE работает поверх обычного HTTP, поэтому прокси-серверы, балансировщики и CDN с ним дружат из коробки. Не нужно настраивать upgrade-запросы, как для WebSockets.

WebSockets в Ktor — для двустороннего общения

Когда понадобился чат, SSE уже не хватало — клиент должен отправлять сообщения. Подключаем плагин WebSockets:

install(WebSockets) {
    pingPeriod = Duration.ofSeconds(15)
    timeout = Duration.ofSeconds(15)
    maxFrameSize = Long.MAX_VALUE
    masking = false
}

val chatRooms = mutableMapOf>()

Обработчик WebSocket-соединения:

routing {
    webSocket("/ws/chat/{roomId}") {
        val roomId = call.parameters["roomId"] ?: return@webSocket close(
            CloseReason(CloseReason.Codes.PROTOCOL_ERROR, "Missing roomId")
        )

        val room = chatRooms.getOrPut(roomId) { mutableListOf() }
        room.add(this)

        try {
            send(Json.encodeToString(ChatMessage(
                type = "system",
                text = "Оператор подключился к чату"
            )))

            for (frame in incoming) {
                when (frame) {
                    is Frame.Text -> {
                        val message = Json.decodeFromString(frame.readText())
                        chatService.saveMessage(roomId.toLong(), message)

                        val outgoing = Json.encodeToString(message)
                        room.forEach { session ->
                            session.send(outgoing)
                        }
                    }
                    is Frame.Close -> {
                        println("Клиент отключился: ${frame.readReason()}")
                    }
                    else -> {}
                }
            }
        } catch (e: Exception) {
            println("WebSocket ошибка: ${e.message}")
        } finally {
            room.remove(this)
        }
    }
}

@Serializable
data class ChatMessage(
    val type: String,
    val text: String,
    val senderId: Long? = null,
    val timestamp: Long = System.currentTimeMillis()
)

Сравнение на практике

После трёх месяцев в проде могу сравнить:

SSE: односторонний (сервер к клиенту), HTTP, автопереподключение через EventSource API в браузере, дружит с любыми прокси, проще в дебаге (видно в DevTools как обычные HTTP-запросы). Минусы — нет бинарных данных, одно соединение равно один поток событий, браузер ограничивает до 6 соединений на домен.

WebSockets: двусторонний, бинарные данные, одно TCP-соединение вместо HTTP-опросов, ниже latency. Минусы — нужен sticky sessions за балансировщиком, сложнее дебажить, не все CDN пропускают, при обрыве переподключение нужно писать вручную.

Мой эвристический правило: если сервер только отправляет данные — SSE. Если клиент тоже отправляет — WebSockets. Исключение: если нужно отправлять бинарные данные (файлы, стриминг) — только WebSockets.

Продакшен-моменты, о которых молчат

Первое — backpressure. Когда мы добавили 50-го подписчика на SSE, канал начал буферизовать события и через минуту получали OutOfMemory. Решение: ограничили размер буфера и при переполнении отключали самых старых подписчиков с HTTP 503.

Второе — мониторинг. В проде нужно знать, сколько активных SSE-соединений и WebSocket-сессий. Мы добавили метрики через Micrometer и дашборд в Grafana. Если соединений вдруг становится ноль — значит, что-то сломалось.

Третье — nginx перед Ktor. Для SSE нужно добавить proxy_buffering off и proxy_cache off, иначе nginx будет буферизовать события и клиент не получит их в реальном времени. Для WebSockets — proxy_set_header Upgrade $http_upgrade и proxy_read_timeout 86400.