Kotlin + gRPC: streaming и двусторонняя связь в продакшене

Полгода назад мы столкнулись с задачей: нужно было доставлять статусы заказов в реальном времени десяткам тысяч клиентов. REST с long polling не справлялся — нагрузка на базу росла экспоненциально, а задержки достигали 5–8 секунд. WebSocket решал проблему, но у нас уже была gRPC-инфраструктура между микросервисами. Решили докрутить её стримингом, и вот что из этого вышло.

Почему gRPC streaming, а не REST или WebSocket

REST по своей природе — request/response. Да, есть Server-Sent Events, но в экосистеме gRPC стриминг работает «из коробки» с автоматическим reconnect, backpressure и типобезопасностью через Protobuf. WebSocket универсален, но у него нет встроенного контракта — вы сами решаете, как описывать сообщения. В gRPC контракт жёстко зафиксирован в .proto-файле, и любые несовместимости выловятся на этапе компиляции.

Для нашего кейса — постоянная доставка обновлений статусов заказов — server-side streaming оказался идеальным. Один запрос от клиента, и сервер непрерывно пушит обновления по одному HTTP/2-соединению.

Описываем контракт в Protobuf

Начнём с proto-файла. Здесь я определил три типа стриминга для разных сценариев:

syntax = "proto3";

package orders.v1;

option java_multiple_files = true;
option java_package = "com.example.orders.grpc";

service OrderStatusService {
  // Server-side streaming: клиент подписывается на обновления
  rpc StreamOrderStatus (StreamOrderStatusRequest)
    returns (stream OrderStatusUpdate);

  // Client-side streaming: загрузка чанков документа
  rpc UploadOrderDocument (stream UploadChunk)
    returns (UploadResponse);

  // Bidirectional: чат поддержки по заказу
  rpc OrderChat (stream ChatMessage)
    returns (stream ChatMessage);
}

message StreamOrderStatusRequest {
  string order_id = 1;
  repeated string event_types = 2;
}

message OrderStatusUpdate {
  string order_id = 1;
  string status = 2;
  int64 timestamp = 3;
  string description = 4;
}

message UploadChunk {
  string order_id = 1;
  string document_type = 2;
  bytes content = 3;
  int32 chunk_index = 4;
  bool is_last = 5;
}

message UploadResponse {
  string document_id = 1;
  int64 total_bytes = 2;
}

message ChatMessage {
  string order_id = 1;
  string sender_id = 2;
  string text = 3;
  int64 timestamp = 4;
}

Server-side streaming: реализация на Kotlin

С библиотекой grpc-kotlin coroutine-стриминг выглядит элегантно. Сервер просто возвращает Flow, а вся магия с HTTP/2 мультиплексированием происходит под капотом:

class OrderStatusServiceImpl(
    private val orderEventRepository: OrderEventRepository,
    private val objectMapper: ObjectMapper
) : OrderStatusServiceGrpcKt.OrderStatusServiceCoroutineImplBase() {

    override suspend fun streamOrderStatus(
        request: StreamOrderStatusRequest
    ): Flow {
        val orderEvents = orderEventRepository.subscribeToOrder(request.orderId)

        return orderEvents
            .filter { event ->
                request.eventTypesList.isEmpty() ||
                event.type in request.eventTypesList
            }
            .map { event ->
                OrderStatusUpdate.newBuilder()
                    .setOrderId(event.orderId)
                    .setStatus(event.status.name)
                    .setTimestamp(event.timestamp.toEpochMilli())
                    .setDescription(event.description)
                    .build()
            }
            .catch { e ->
                logger.error(e) { "Stream error for order ${request.orderId}" }
                throw Status.INTERNAL
                    .withDescription("Stream interrupted")
                    .asRuntimeException()
            }
    }
}

Client-side streaming: загрузка чанками

Когда клиенты загружают документы к заказам, мы используем client-side streaming. Файл бьётся на чанки, и каждый отправляется по отдельности. Это позволяет обрабатывать большие файлы без загрузки всего содержимого в память:

suspend fun uploadDocument(
    orderId: String,
    documentType: String,
    file: Path
): UploadResponse {
    val chunkSize = 64 * 1024 // 64 KB
    val stub = OrderStatusServiceGrpcKt.OrderStatusServiceCoroutineStub(channel)

    val chunkFlow = flow {
        file.inputStream().buffered().use { input ->
            var index = 0
            val buffer = ByteArray(chunkSize)
            var bytesRead: Int

            while (input.read(buffer).also { bytesRead = it } != -1) {
                val isLast = bytesRead < chunkSize
                emit(
                    UploadChunk.newBuilder()
                        .setOrderId(orderId)
                        .setDocumentType(documentType)
                        .setContent(
                            ByteString.copyFrom(buffer, 0, bytesRead)
                        )
                        .setChunkIndex(index++)
                        .setIsLast(isLast)
                        .build()
                )
            }
        }
    }

    return stub.uploadOrderDocument(chunkFlow)
}

Bidirectional streaming: чат поддержки

Двусторонний стриминг мы задействовали для чата поддержки внутри заказа. Обе стороны шлют сообщения, и обе получают. Реализация на корутинах выглядит как два параллельных потока:

override suspend fun orderChat(requests: Flow): Flow {
    val responses = MutableSharedFlow()

    // Обрабатываем входящие сообщения в отдельной корутине
    val incomingJob = launch {
        requests.collect { message ->
            val processed = handleMessage(message)
            responses.emit(processed)

            // Если это сообщение от клиента — отправляем автоответ
            if (message.senderId.startsWith("client_")) {
                val autoReply = generateAutoReply(message)
                chatRepository.save(message.orderId, autoReply)
                responses.emit(autoReply)
            }
        }
    }

    return responses.onCompletion {
        incomingJob.cancel()
    }
}

Backpressure и flow control

Одна из главных проблем, с которой мы столкнулись — backpressure. Когда клиент не успевает обрабатывать сообщения, буфер HTTP/2 растёт, и в какой-то момент connection reset. Решение — использовать .conflate() для некритичных обновлений (оставляем только последнее) и явный .buffer() с ограничением размера для важных:

// Для статусов — пропускаем промежуточные, важен только актуальный
orderEvents.conflate().map { ... }

// Для чата — буферизуем, но с лимитом
chatMessages.buffer(onBufferOverflow = BufferOverflow.SUSPEND)

Обработка ошибок в стримах

В стриминге ошибка не заканчивает HTTP-соединение автоматически — нужно явно отправить статус через gRPC. Мы обернули все стримы в .catch {} и отправляем клиенту понятный статус. Отдельно обрабатываем DEADLINE_EXCEEDED — если клиент не отправлял heartbeats 30 секунд, закрываем стрим с причиной.

Балансировка: ловушка стриминговых соединений

Это был самый больной момент. L4-балансировщики (HAProxy, Envoy по умолчанию) распределяют новые соединения, но gRPC стримы живут минуты и часы. В итоге первый подхватил 70% соединений, остальные простаивали. Решение: перешли на L7 с policy ROUND_ROBIN на уровне запросов, а не соединений. В Envoy это grpc_lb с RING_HASH.

Итоги после полугода в проде

gRPC streaming оказался правильным выбором для нашей задачи. Средняя задержка доставки статуса упала с 6 секунд (long polling) до 120 мс. Потребление памяти на инстансе сократилось в 3 раза по сравнению с WebSocket-решением на Spring. Но будьте готовы потратить время на настройку балансировки и мониторинг — стриминг требует внимания к деталям, которые в request/response просто не возникают.