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

Введение: за рамками простого WebSocket на Go

Базовые примеры WebSocket на Go обычно показывают простое эхо-соединение. В реальных продакшн-проектах — от биржевых терминалов до многопользовательских игр и систем оповещений — требуется гораздо больше: управление комнатами, Graceful Shutdown, масштабирование и корректная обработка ошибок. В этой статье мы разберем продвинутый паттерн построения real-time приложений на Go с использованием библиотеки gorilla/websocket.

Мы создадим архитектуру на основе Hub (центральный диспетчер) и Client (индивидуальное соединение), добавим поддержку комнат, механизм отказоустойчивости и покажем, как правильно завершать работу сервера.

Архитектура Hub и Client: основа real-time сервера

Почему Hub?

Каждый WebSocket-клиент — это отдельная горутина. Управлять ими напрямую из HTTP-обработчика опасно: это приводит к гонкам данных и утечкам памяти. Паттерн Hub решает проблему, выступая единым каналом для регистрации, удаления и рассылки сообщений.

Реализация Hub

package main

import (
    "log"
    "sync"
)

type Client struct {
    hub  *Hub
    send chan []byte
    room string
}

type Hub struct {
    clients    map[*Client]bool
    broadcast  chan Message
    register   chan *Client
    unregister chan *Client
    mu         sync.RWMutex
}

type Message struct {
    Room    string
    Data    []byte
    Sender  *Client
}

func NewHub() *Hub {
    return &Hub{
        clients:    make(map[*Client]bool),
        broadcast:  make(chan Message, 256),
        register:   make(chan *Client),
        unregister: make(chan *Client),
    }
}

func (h *Hub) Run() {
    for {
        select {
        case client := <-h.register:
            h.mu.Lock()
            h.clients[client] = true
            h.mu.Unlock()
            log.Printf("Клиент подключен. Всего: %d", len(h.clients))

        case client := <-h.unregister:
            h.mu.Lock()
            if _, ok := h.clients[client]; ok {
                delete(h.clients, client)
                close(client.send)
            }
            h.mu.Unlock()

        case message := <-h.broadcast:
            h.mu.RLock()
            for client := range h.clients {
                // Отправляем только в нужную комнату
                if client.room == message.Room || message.Room == "" {
                    select {
                    case client.send <- message.Data:
                    default:
                        // Если канал переполнен — отключаем клиента
                        h.mu.RUnlock()
                        h.unregister <- client
                        h.mu.RLock()
                    }
                }
            }
            h.mu.RUnlock()
        }
    }
}

Ключевые моменты:

• Каналы — register, unregister, broadcast — безопасно синхронизируют горутины.
• RWMutex — позволяет одновременно читать список клиентов, но блокирует запись.
• Room-based routing — сообщение отправляется только клиентам из той же комнаты.

Управление комнатами и подписками

Комнаты (Rooms) — это логические группы клиентов. Например, в чате у каждой беседы своя комната. В нашем Hub комната задается полем room у Client. Рассмотрим, как клиент подписывается на комнату.

HTTP-обработчик с WebSocket Upgrade

import (
    "net/http"
    "github.com/gorilla/websocket"
)

var upgrader = websocket.Upgrader{
    ReadBufferSize:  1024,
    WriteBufferSize: 1024,
    CheckOrigin: func(r *http.Request) bool {
        return true // Для продакшена заменить на проверку origin
    },
}

func serveWs(hub *Hub, w http.ResponseWriter, r *http.Request) {
    conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        log.Println("Ошибка upgrade:", err)
        return
    }

    room := r.URL.Query().Get("room")
    if room == "" {
        room = "default"
    }

    client := &Client{
        hub:  hub,
        send: make(chan []byte, 256),
        room: room,
    }

    hub.register <- client

    // Запускаем горутины для чтения и записи
    go client.writePump(conn)
    go client.readPump(conn)
}

ReadPump и WritePump

func (c *Client) readPump(conn *websocket.Conn) {
    defer func() {
        c.hub.unregister <- c
        conn.Close()
    }()

    for {
        _, message, err := conn.ReadMessage()
        if err != nil {
            if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
                log.Printf("Ошибка чтения: %v", err)
            }
            break
        }

        // Рассылаем сообщение в комнату клиента
        c.hub.broadcast <- Message{
            Room:   c.room,
            Data:   message,
            Sender: c,
        }
    }
}

func (c *Client) writePump(conn *websocket.Conn) {
    ticker := time.NewTicker(30 * time.Second) // Ping каждые 30 сек
    defer func() {
        ticker.Stop()
        conn.Close()
    }()

    for {
        select {
        case message, ok := <-c.send:
            if !ok {
                conn.WriteMessage(websocket.CloseMessage, []byte{})
                return
            }
            if err := conn.WriteMessage(websocket.TextMessage, message); err != nil {
                return
            }

        case <-ticker.C:
            conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
            if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil {
                return
            }
        }
    }
}

Важно: SetWriteDeadline и PingMessage предотвращают "мертвые" соединения. Без этого клиент может висеть часами, потребляя ресурсы.

Graceful Shutdown и масштабирование

Корректное завершение работы

При остановке сервера (SIGINT, SIGTERM) нужно закрыть все WebSocket-соединения штатно, отправив клиентам код закрытия. Иначе клиенты будут ждать ответа вечно.

import (
    "context"
    "net/http"
    "os/signal"
    "syscall"
)

func main() {
    hub := NewHub()
    go hub.Run()

    http.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
        serveWs(hub, w, r)
    })

    server := &http.Server{Addr: ":8080"}

    // Graceful shutdown
    ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM