WebSocket на Kotlin: создание real-time приложений продвинутого уровня

Введение: Зачем нужен продвинутый WebSocket на Kotlin?

Современные веб-приложения всё чаще требуют мгновенной передачи данных: чаты, уведомления, онлайн-игры, финансовые котировки и панели мониторинга. Базовый протокол WebSocket позволяет установить постоянное двунаправленное соединение между клиентом и сервером, но на практике разработка надежного и масштабируемого real-time приложения на Kotlin требует гораздо большего, чем просто открытие сокета.

В этой статье мы выйдем за рамки простого эхо-сервера. Мы рассмотрим архитектурные паттерны для работы с тысячами соединений, обработку ошибок с использованием корутин Kotlin, кластеризацию серверов с брокерами сообщений и вопросы безопасности. Материал рассчитан на разработчиков, уже знакомых с основами Kotlin и Ktor.

1. Архитектура высоконагруженного WebSocket сервера на Ktor

Для production-решения недостаточно просто повесить обработчик webSocket. Необходимо продумать управление сессиями, жизненный цикл соединения и изоляцию обработчиков. Рассмотрим ключевые элементы.

Управление сессиями и их группировка

Каждое WebSocket-соединение в Ktor представлено объектом DefaultWebSocketSession. Для отправки сообщений конкретному пользователю или группе (например, "комната чата") нужно хранить активные сессии. Используйте ConcurrentHashMap для потокобезопасности.

import io.ktor.websocket.*
import java.util.concurrent.ConcurrentHashMap

class SessionManager {
    private val sessions = ConcurrentHashMap<String, WebSocketSession>()
    private val roomSessions = ConcurrentHashMap<String, MutableSet<String>>()

    fun addUserToRoom(roomId: String, userId: String, session: WebSocketSession) {
        sessions[userId] = session
        roomSessions.computeIfAbsent(roomId) { mutableSetOf() }.add(userId)
    }

    suspend fun broadcastToRoom(roomId: String, message: String) {
        roomSessions[roomId]?.forEach { userId ->
            sessions[userId]?.send(Frame.Text(message))
        }
    }

    fun removeUser(userId: String) {
        sessions.remove(userId)
        roomSessions.forEach { (_, users) -> users.remove(userId) }
    }
}

Обработка жизненного цикла с корутинами

Ktor использует корутины для обработки каждого соединения. Важно корректно обрабатывать исключения и закрытие сокета, чтобы избежать утечек памяти. Используйте try-catch внутри блока webSocket и вызывайте removeUser в блоке finally.

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

embeddedServer(Netty, port = 8080) {
    val sessionManager = SessionManager()
    routing {
        webSocket("/chat/{roomId}") {
            val roomId = call.parameters["roomId"] ?: "general"
            val userId = call.request.queryParameters["userId"] ?: "anon"
            sessionManager.addUserToRoom(roomId, userId, this)
            try {
                for (frame in incoming) {
                    if (frame is Frame.Text) {
                        val text = frame.readText()
                        sessionManager.broadcastToRoom(roomId, "[$userId]: $text")
                    }
                }
            } catch (e: Exception) {
                log.error("WebSocket error for user $userId: ${e.localizedMessage}")
            } finally {
                sessionManager.removeUser(userId)
                log.info("User $userId disconnected from room $roomId")
            }
        }
    }
}.start(wait = true)

2. Масштабирование и кластеризация WebSocket на Kotlin

Один сервер не справится с десятками тысяч одновременных соединений. Решение — горизонтальное масштабирование. Однако WebSocket-соединение "привязано" к конкретному экземпляру сервера. Как отправить сообщение пользователю, подключенному к другому серверу?

Брокер сообщений (Redis Pub/Sub)

Классический подход — использовать Redis Pub/Sub или RabbitMQ. Каждый экземпляр сервера подписывается на канал и публикует в него сообщения, предназначенные для других инстансов.

import redis.clients.jedis.Jedis
import kotlinx.coroutines.*

class RedisPubSub(private val sessionManager: SessionManager) {
    private val jedis = Jedis("localhost")
    private val pubSub = object : JedisPubSub() {
        override fun onMessage(channel: String, message: String) {
            // Запускаем в корутине, чтобы не блокировать поток Redis
            GlobalScope.launch {
                sessionManager.broadcastToRoom(channel, message)
            }
        }
    }

    fun subscribeToRoom(roomId: String) {
        jedis.subscribe(pubSub, roomId)
    }

    fun publishToRoom(roomId: String, message: String) {
        jedis.publish(roomId, message)
    }
}

Важно: Подписка на Redis блокирует поток, поэтому её нужно выполнять в отдельном корутиновом контексте (например, Dispatchers.IO).

Sticky Sessions vs. Stateful Routing

В кластерной среде необходимо перенаправлять запросы одного пользователя на тот же сервер. Для этого используйте:

• Sticky sessions (липкие сессии) — балансировщик (например, Nginx) запоминает сервер по куке или IP.
• Stateful routing — внешнее хранилище (Redis/ZooKeeper) хранит маппинг: userId → serverId. Балансировщик или шлюз (Kong, Envoy) использует эти данные.

Для high-availability рекомендуется второй подход, так как он не теряет данные при падении одного сервера (сессию можно переподнять на другом).

3. Продвинутая обработка ошибок и восстановление соединений

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

Механизм Heartbeat и таймауты

Настройте pingPeriod в Ktor, чтобы сервер отправлял пинг-понги. Если клиент не отвечает в течение timeout, соединение закрывается. На стороне клиента (JavaScript) реализуйте экспоненциальную задержку переподключения.

// Kotlin сервер: настройка уже показана выше
// Пример клиента на JS:
let reconnectDelay = 1000;
function connect() {
    const ws = new WebSocket("ws://localhost:8080