> Как организовать запись данных из WebSocket в базу данных и кэш (Go)

Короткий ответ

Основная задача - не писать в БД и кэш синхронно в обработчике WebSocket-сообщения, а разделить приём данных, их валидацию, буферизацию и асинхронную запись. Используем паттерн "pipeline": WebSocket handler → канал/очередь → worker pool → batch insert в БД и обновление кэша. Кэш обновляем через write-through или write-behind, инвалидацию делаем после успешного коммита в БД. Для надёжности добавляем retry, idempotency key и graceful shutdown.

Подробное объяснение

WebSocket - это долгоживущее соединение с высокой частотой сообщений. Синхронная запись каждого сообщения в БД создаст узкое место и заблокирует чтение из сокета. Поэтому применяем асинхронную архитектуру.

Основные компоненты:

1. Handler - принимает сообщение, парсит, валидирует, кладёт в буферизованный канал (или отправляет в очередь типа NATS, RabbitMQ, Kafka, если нужна горизонтальная масштабируемость).
2. Worker pool - N горутин читают из канала, группируют сообщения в батчи (например, по 100 штук или по таймауту 100ms) и выполняют bulk insert.
3. БД - используем INSERT ... ON CONFLICT DO UPDATE или batch insert в транзакции. Это снижает количество round-trip.
4. Кэш - стратегия зависит от сценария чтения:
   • write-through: пишем в кэш сразу после успешной записи в БД;
   • write-behind: пишем в кэш асинхронно, с задержкой;
   • cache-aside: обновляем кэш только при чтении, инвалидируя ключ после записи. Для WebSocket чаще подходит write-through с инвалидацией по ключу, чтобы клиенты сразу видели актуальные данные.
5. Идемпотентность - каждое сообщение должно иметь уникальный ID (например, UUID или sequence number), чтобы повторная обработка не создавала дубликаты.
6. Обработка ошибок - если запись в БД не удалась, кладём сообщение обратно в очередь с retry и экспоненциальным backoff. После N попыток - в DLQ (dead letter queue) или лог.
7. Backpressure - если канал переполнен, нужно либо блокировать чтение из сокета, либо отклонять сообщения с ошибкой 429. Иначе память будет исчерпана.
8. Graceful shutdown - при завершении приложения дожидаемся, пока все сообщения из канала будут обработаны, или сохраняем их в persistent queue.

На практике

Для Go-приложения типичная схема:

• Канал с буфером, например, 1024 сообщения.
• 4-8 воркеров (число можно регулировать через runtime.GOMAXPROCS или нагрузочное тестирование).
• Батчинг: воркер накапливает сообщения в слайсе, пока не наберёт 100 штук или не пройдёт 200ms - что раньше.
• Транзакция: BEGIN; INSERT ...; COMMIT; - если одно сообщение из батча невалидно, откатываем весь батч и обрабатываем по одному.
• Кэш (Redis): после коммита делаем SET key value с TTL, или DEL key если данные читаются через отдельный запрос.
• Для распределённой системы - используем Redis Streams или Kafka вместо канала, это даёт гарантии доставки и возможность нескольким инстансам обрабатывать сообщения.

Пример кода

GO
type Message struct {
    ID   string `json:"id"`
    Data []byte `json:"data"`
}

type Handler struct {
    ch       chan Message
    workers  int
    batchSize int
    flushTimeout time.Duration
    db       *sql.DB
    cache    *redis.Client
}

func NewHandler(db *sql.DB, cache *redis.Client) *Handler {
    h := &Handler{
        ch:          make(chan Message, 1024),
        workers:     4,
        batchSize:   100,
        flushTimeout: 200 * time.Millisecond,
        db:          db,
        cache:       cache,
    }
    for i := 0; i < h.workers; i++ {
        go h.worker()
    }
    return h
}

func (h *Handler) HandleWS(conn *websocket.Conn) {
    for {
        _, data, err := conn.ReadMessage()
        if err != nil {
            return
        }
        var msg Message
        if err := json.Unmarshal(data, &msg); err != nil {
            continue // или отправить ошибку клиенту
        }
        select {
        case h.ch <- msg:
        default:
            // backpressure: переполнен канал
            conn.WriteMessage(websocket.TextMessage, []byte(`{"error":"too many requests"}`))
        }
    }
}

func (h *Handler) worker() {
    batch := make([]Message, 0, h.batchSize)
    timer := time.NewTimer(h.flushTimeout)
    defer timer.Stop()

    flush := func() {
        if len(batch) == 0 {
            return
        }
        h.writeBatch(batch)
        batch = batch[:0]
        timer.Reset(h.flushTimeout)
    }

    for {
        select {
        case msg := <-h.ch:
            batch = append(batch, msg)
            if len(batch) >= h.batchSize {
                flush()
            }
        case <-timer.C:
            flush()
        }
    }
}

func (h *Handler) writeBatch(messages []Message) {
    tx, err := h.db.Begin()
    if err != nil {
        // retry logic
        return
    }
    defer tx.Rollback()

    stmt, _ := tx.Prepare(`INSERT INTO messages (id, data) VALUES ($1, $2) ON CONFLICT (id) DO UPDATE SET data = EXCLUDED.data`)
    for _, m := range messages {
        if _, err := stmt.Exec(m.ID, m.Data); err != nil {
            return // откат транзакции
        }
    }
    if err := tx.Commit(); err != nil {
        return
    }

    // write-through кэш
    for _, m := range messages {
        h.cache.Set(context.Background(), "msg:"+m.ID, m.Data, 10*time.Minute)
    }
}

Как отвечать на собеседовании

Начни с главной проблемы - синхронная запись блокирует WebSocket. Затем опиши архитектуру: канал, воркеры, батчинг. Упомяни trade-off между каналом в памяти и внешней очередью. Обязательно скажи про идемпотентность и обработку ошибок. Если спросят про масштабирование - предложи Kafka или Redis Streams. Покажи понимание backpressure и graceful shutdown. Не углубляйся в детали реализации, если не просят - сначала дай общую картину.

Что проверяет интервьюер

• Понимание асинхронной обработки и паттернов конкурентности в Go.
• Умение проектировать отказоустойчивые системы: retry, идемпотентность, DLQ.
• Знание стратегий кэширования и их trade-off.
• Понимание backpressure и ограничения ресурсов.
• Способность объяснить, почему батчинг эффективнее одиночных записей.
• Внимание к graceful shutdown и потере данных.

Типичные ошибки

• Синхронная запись в БД внутри обработчика WebSocket.
• Неограниченный буфер канала - приводит к OOM.
• Игнорирование ошибок записи - потеря данных.
• Отсутствие идемпотентности - дубликаты при retry.
• Обновление кэша до коммита в БД - рассинхронизация.
• Нет обработки переполнения канала - клиент получает блокировку.
• Забывают про закрытие соединения и drain канала при shutdown.