xDev Asia

ゴリラ/WebSocket、接続ライフサイクル。チャットルーム、ブロードキャストパターン。リアルタイム通知、ハートビート、再接続。 WebSocket のスケーリング。

Golang: 基本から高度まで

パート 4: 高度な機能

xdev.asia

1. WebSocketの基礎

WebSocket は、単一の TCP 接続を介してクライアントとサーバー間の全二重接続を提供します。 HTTP リクエスト/レスポンスとは異なり、WebSocket を使用すると、サーバーはプロアクティブにデータをクライアントにプッシュできます。

特長	HTTP	Webソケット
接続。接続	短命	永続的
方向	クライアント → サーバー	双方向
プロトコル	HTTP/1.1、HTTP/2	ws://、wss://
オーバーヘッド	リクエストごとのヘッダー	最小限のフレームヘッダー
ユースケース	REST API	チャット、通知、ライブデータ

2. ゴリラ/ウェブソケットのセットアップ

go get github.com/gorilla/websocket

2.1.基本的なエコーサーバー

package main

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

var upgrader = websocket.Upgrader{
    ReadBufferSize:  1024,
    WriteBufferSize: 1024,
    CheckOrigin: func(r *http.Request) bool {
        // Production: validate origin
        origin := r.Header.Get("Origin")
        return origin == "https://example.com"
    },
}

func echoHandler(w http.ResponseWriter, r *http.Request) {
    conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        log.Printf("Upgrade error: %v", err)
        return
    }
    defer conn.Close()
    
    for {
        messageType, message, err := conn.ReadMessage()
        if err != nil {
            if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
                log.Printf("Read error: %v", err)
            }
            break
        }
        
        log.Printf("Received: %s", message)
        
        if err := conn.WriteMessage(messageType, message); err != nil {
            log.Printf("Write error: %v", err)
            break
        }
    }
}

func main() {
    http.HandleFunc("/ws", echoHandler)
    log.Fatal(http.ListenAndServe(":8080", nil))
}

3. チャットルーム — ハブパターン

3.1.ハブ (接続管理)

package ws

import (
    "encoding/json"
    "log"
    "sync"
)

type Message struct {
    Type    string `json:"type"`
    Room    string `json:"room"`
    Sender  string `json:"sender"`
    Content string `json:"content"`
}

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

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

func (h *Hub) Run() {
    for {
        select {
        case client := <-h.register:
            h.mu.Lock()
            if h.rooms[client.Room] == nil {
                h.rooms[client.Room] = make(map[*Client]bool)
            }
            h.rooms[client.Room][client] = true
            h.mu.Unlock()
            
            log.Printf("Client joined room %s (total: %d)", client.Room, len(h.rooms[client.Room]))
            
        case client := <-h.unregister:
            h.mu.Lock()
            if clients, ok := h.rooms[client.Room]; ok {
                if _, ok := clients[client]; ok {
                    delete(clients, client)
                    close(client.Send)
                    if len(clients) == 0 {
                        delete(h.rooms, client.Room)
                    }
                }
            }
            h.mu.Unlock()
            
        case msg := <-h.broadcast:
            h.mu.RLock()
            clients := h.rooms[msg.Room]
            h.mu.RUnlock()
            
            data, _ := json.Marshal(msg)
            for client := range clients {
                select {
                case client.Send <- data:
                default:
                    // Client buffer full, disconnect
                    close(client.Send)
                    h.mu.Lock()
                    delete(h.rooms[msg.Room], client)
                    h.mu.Unlock()
                }
            }
        }
    }
}

3.2.クライアント

package ws

import (
    "encoding/json"
    "log"
    "time"
    
    "github.com/gorilla/websocket"
)

const (
    writeWait      = 10 * time.Second
    pongWait       = 60 * time.Second
    pingPeriod     = (pongWait * 9) / 10
    maxMessageSize = 4096
)

type Client struct {
    Hub    *Hub
    Conn   *websocket.Conn
    Send   chan []byte
    Room   string
    UserID string
}

// ReadPump: đọc messages từ WebSocket connection
func (c *Client) ReadPump() {
    defer func() {
        c.Hub.unregister <- c
        c.Conn.Close()
    }()
    
    c.Conn.SetReadLimit(maxMessageSize)
    c.Conn.SetReadDeadline(time.Now().Add(pongWait))
    c.Conn.SetPongHandler(func(string) error {
        c.Conn.SetReadDeadline(time.Now().Add(pongWait))
        return nil
    })
    
    for {
        _, data, err := c.Conn.ReadMessage()
        if err != nil {
            if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
                log.Printf("Read error: %v", err)
            }
            break
        }
        
        var msg Message
        if err := json.Unmarshal(data, &msg); err != nil {
            continue
        }
        
        msg.Sender = c.UserID
        msg.Room = c.Room
        c.Hub.broadcast <- msg
    }
}

// WritePump: gửi messages qua WebSocket connection
func (c *Client) WritePump() {
    ticker := time.NewTicker(pingPeriod)
    defer func() {
        ticker.Stop()
        c.Conn.Close()
    }()
    
    for {
        select {
        case message, ok := <-c.Send:
            c.Conn.SetWriteDeadline(time.Now().Add(writeWait))
            if !ok {
                c.Conn.WriteMessage(websocket.CloseMessage, []byte{})
                return
            }
            
            w, err := c.Conn.NextWriter(websocket.TextMessage)
            if err != nil {
                return
            }
            w.Write(message)
            
            // Batch queued messages
            n := len(c.Send)
            for i := 0; i < n; i++ {
                w.Write([]byte("\n"))
                w.Write(<-c.Send)
            }
            
            if err := w.Close(); err != nil {
                return
            }
            
        case <-ticker.C:
            c.Conn.SetWriteDeadline(time.Now().Add(writeWait))
            if err := c.Conn.WriteMessage(websocket.PingMessage, nil); err != nil {
                return
            }
        }
    }
}

3.3. Jin を使用した WebSocket ハンドラー

func WebSocketHandler(hub *Hub) gin.HandlerFunc {
    return func(c *gin.Context) {
        room := c.Param("room")
        userID := c.GetString("user_id") // from auth middleware
        
        conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
        if err != nil {
            log.Printf("Upgrade error: %v", err)
            return
        }
        
        client := &Client{
            Hub:    hub,
            Conn:   conn,
            Send:   make(chan []byte, 256),
            Room:   room,
            UserID: userID,
        }
        
        hub.register <- client
        
        go client.WritePump()
        go client.ReadPump()
    }
}

// Router
func SetupWSRoutes(r *gin.Engine, hub *Hub) {
    r.GET("/ws/:room", AuthMiddleware(), WebSocketHandler(hub))
}

4. ハートビートと再接続

// Client-side (JavaScript)
class WSClient {
    constructor(url) {
        this.url = url;
        this.reconnectDelay = 1000;
        this.maxReconnectDelay = 30000;
        this.connect();
    }
    
    connect() {
        this.ws = new WebSocket(this.url);
        
        this.ws.onopen = () => {
            console.log('Connected');
            this.reconnectDelay = 1000; // Reset delay
        };
        
        this.ws.onclose = (event) => {
            if (!event.wasClean) {
                // Exponential backoff reconnection
                setTimeout(() => this.connect(), this.reconnectDelay);
                this.reconnectDelay = Math.min(
                    this.reconnectDelay * 2,
                    this.maxReconnectDelay
                );
            }
        };
        
        this.ws.onmessage = (event) => {
            const msg = JSON.parse(event.data);
            this.handleMessage(msg);
        };
    }
    
    send(type, content) {
        if (this.ws.readyState === WebSocket.OPEN) {
            this.ws.send(JSON.stringify({ type, content }));
        }
    }
}

5. Redis Pub/Sub を使用した WebSocket のスケーリング

import "github.com/redis/go-redis/v9"

type RedisHub struct {
    *Hub
    redis *redis.Client
}

func (h *RedisHub) Subscribe(ctx context.Context, room string) {
    sub := h.redis.Subscribe(ctx, "ws:"+room)
    ch := sub.Channel()
    
    for msg := range ch {
        var wsMsg Message
        json.Unmarshal([]byte(msg.Payload), &wsMsg)
        
        // Broadcast to local clients only
        h.mu.RLock()
        clients := h.rooms[room]
        h.mu.RUnlock()
        
        data, _ := json.Marshal(wsMsg)
        for client := range clients {
            select {
            case client.Send <- data:
            default:
                close(client.Send)
            }
        }
    }
}

func (h *RedisHub) Publish(ctx context.Context, msg Message) error {
    data, _ := json.Marshal(msg)
    return h.redis.Publish(ctx, "ws:"+msg.Room, data).Err()
}

次の記事: gRPC とプロトコルバッファー — 高性能 RPC 通信。

レッスン 13: WebSocket とリアルタイム