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

WebSocket на Go: продвинутые техники для real-time приложений

Разработка real-time приложений на Go — одна из сильнейших сторон языка. Встроенная поддержка конкурентности через горутины и каналы делает Go идеальным выбором для WebSocket-серверов. Однако, когда речь заходит о продакшн-системах с тысячами одновременных соединений, простой пример "Hello, WebSocket" перестаёт работать. В этой статье мы разберём продвинутые техники: от оптимизации пула соединений до масштабирования через брокеры сообщений.

1. Архитектура пула соединений: управление тысячами клиентов

В базовом примере мы создаём отдельную горутину на каждое соединение. Это нормально для 100-200 клиентов, но при 10 000+ соединений расход памяти и накладные расходы на переключение контекста становятся критическими. Продвинутый подход — использование пула соединений с фиксированным числом воркеров и каналов.

// Pool — пул соединений с ограничением горутин
type Pool struct {
    connections map[*websocket.Conn]bool
    broadcast   chan []byte
    register    chan *websocket.Conn
    unregister  chan *websocket.Conn
}

func NewPool(maxGoroutines int) *Pool {
    return &Pool{
        connections: make(map[*websocket.Conn]bool),
        broadcast:   make(chan []byte, 256),
        register:    make(chan *websocket.Conn),
        unregister:  make(chan *websocket.Conn),
    }
}

func (pool *Pool) Start() {
    for {
        select {
        case conn := <-pool.register:
            pool.connections[conn] = true
        case conn := <-pool.unregister:
            if _, ok := pool.connections[conn]; ok {
                delete(pool.connections, conn)
                conn.Close()
            }
        case message := <-pool.broadcast:
            for conn := range pool.connections {
                go func(c *websocket.Conn) {
                    if err := c.WriteMessage(websocket.TextMessage, message); err != nil {
                        pool.unregister <- c
                    }
                }(conn)
            }
        }
    }
}

Почему это работает? Каналы действуют как буферы, предотвращая блокировку. Один селект управляет всеми событиями, а горутины для записи создаются только при необходимости. Это снижает пиковую нагрузку на GC (сборщик мусора).

2. Graceful shutdown и управление состоянием соединений

При перезапуске сервера важно корректно закрыть все WebSocket-соединения, отправив клиентам сообщение о завершении работы. Игнорирование этого шага приводит к "висячим" соединениям и утечкам памяти.

// Server — WebSocket сервер с graceful shutdown
type Server struct {
    pool    *Pool
    httpSrv *http.Server
}

func (s *Server) Shutdown(ctx context.Context) error {
    // 1. Закрываем HTTP-сервер (новые соединения не принимаются)
    if err := s.httpSrv.Shutdown(ctx); err != nil {
        return err
    }

    // 2. Отправляем всем клиентам сообщение о завершении
    shutdownMsg := []byte(`{"type":"shutdown","message":"server is going down"}`)
    s.pool.broadcast <- shutdownMsg

    // 3. Закрываем все соединения в пуле
    for conn := range s.pool.connections {
        conn.WriteMessage(websocket.CloseMessage, 
            websocket.FormatCloseMessage(websocket.CloseNormalClosure, "server shutdown"))
        conn.Close()
    }

    return nil
}

// Использование в main
func main() {
    pool := NewPool(1000)
    go pool.Start()

    srv := &Server{pool: pool}
    srv.httpSrv = &http.Server{Addr: ":8080"}

    // Обработка сигналов
    go func() {
        sigint := make(chan os.Signal, 1)
        signal.Notify(sigint, os.Interrupt, syscall.SIGTERM)
        <-sigint
        ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
        defer cancel()
        srv.Shutdown(ctx)
    }()

    log.Fatal(srv.httpSrv.ListenAndServe())
}

Важно: Всегда используйте context.WithTimeout для shutdown — это предотвращает бесконечное ожидание закрытия "залипших" соединений.

3. Масштабирование через брокер сообщений (Redis Pub/Sub)

Когда одного экземпляра Go-сервера недостаточно, требуется горизонтальное масштабирование. Проблема: клиенты, подключённые к разным серверам, не могут получать сообщения друг от друга. Решение — внешний брокер, например Redis Pub/Sub.

// RedisHub — мост между WebSocket серверами через Redis
type RedisHub struct {
    redisClient *redis.Client
    pool        *Pool
    channelName string
}

func NewRedisHub(addr string, pool *Pool) *RedisHub {
    rdb := redis.NewClient(&redis.Options{
        Addr: addr,
        PoolSize: 100, // пул соединений к Redis
    })
    return &RedisHub{
        redisClient: rdb,
        pool:        pool,
        channelName: "websocket:messages",
    }
}

// Listen — подписка на канал Redis
func (h *RedisHub) Listen(ctx context.Context) {
    pubsub := h.redisClient.Subscribe(ctx, h.channelName)
    defer pubsub.Close()

    ch := pubsub.Channel()
    for msg := range ch {
        // Полученное сообщение отправляем всем локальным клиентам
        h.pool.broadcast <- []byte(msg.Payload)
    }
}

// Publish — публикация сообщения от локального клиента
func (h *RedisHub) Publish(ctx context.Context, message []byte) error {
    return h.redisClient.Publish(ctx, h.channelName, string(message)).Err()
}

Альтернативы Redis: NATS (лёгкий и быстрый), RabbitMQ (гарантированная доставка) или Apache Kafka (для high-throughput систем). Выбор зависит от требований к надёжности и задержкам.

4. Оптимизация производительности: сжатие, буферизация и лимиты

Для real-time приложений критична минимальная задержка. Три простых, но эффективных приёма:

• Сжатие WebSocket (permessage-deflate) — уменьшает размер сообщений на 40-60%, но добавляет нагрузку на CPU. Включается при handshake:

var upgrader = websocket.Upgrader{
    EnableCompression: true,
    ReadBufferSize:    1024,
    WriteBufferSize:   1024,
}

• Буферизация записи — вместо отправки каждого сообщения сразу, накапливайте их и отправляйте пачками:

// BufferedWriter — буферизированная запись с таймаутом
type BufferedWriter struct {
    conn   *websocket.Conn