Go WebSocketリアルタイムチャット実戦：ゼロからスケーラブルなメッセージサーバーを構築する5つのキーアーキテクチャ

Go WebSocketリアルタイムチャット：2026年にまだポーリングを使っていますか？

まだHTTPポーリングでリアルタイムメッセージングを実装していますか？毎秒100リクエストでサーバーが耐えられず、メッセージ遅延が3-5秒？WebSocketは全二重リアルタイム通信の標準プロトコルです。Go言語のgoroutineモデルは高同時WebSocketサービスに天然適しています。2026年、gorilla/websocketとnhooyr.io/websocketの2大ライブラリが成熟し、Redis Pub/SubやKafkaと組み合わせることで、10万+同時接続をサポートするリアルタイムチャットシステムを簡単に構築できます。

本記事では5つのキーアーキテクチャから、接続管理→ルームモデル→メッセージブロードキャスト→永続化ストレージ→水平スケーリングのフルパイプライン実戦を解説します。

---

コア概念

概念	説明
WebSocket	全二重永続接続プロトコル、RFC 6455
Hub	接続管理センター、すべてのクライアントを維持
Room	チャットルーム、メッセージブロードキャストの分離単位
Client	WebSocketクライアント接続ラッパー
Pub/Sub	パブリッシュ・サブスクライブパターン、プロセス間メッセージ配信に使用
Backpressure	背圧制御、遅いクライアントがサーバーを圧迫するのを防止
Heartbeat	ハートビート検出、切断を迅速に発見
Message Queue	メッセージキュー、メッセージの確実な配信を保証

---

問題分析：WebSocketチャットの5つの課題

1. 接続管理：10万+接続のメモリとgoroutineオーバーヘッド
2. メッセージブロードキャスト：1つのメッセージをルーム内の全オンラインユーザーにプッシュ
3. 遅いクライアント：消費が遅いクライアントがサーバーのメモリオーバーフローを引き起こす
4. 水平スケーリング：単一マシンで対応できず、複数インスタンスの連携が必要
5. メッセージ永続化：オフラインユーザーがオンライン時に履歴メッセージを取得する必要

---

ステップバイステップ：5つのWebSocketチャットコアアーキテクチャ

アーキテクチャ1：Hub-Client接続管理モデル

// internal/chat/client.go
package chat

import (
	"encoding/json"
	"log"
	"time"

	"github.com/gorilla/websocket"
)

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

type Client struct {
	hub      *Hub
	conn     *websocket.Conn
	send     chan []byte
	userID   string
	username string
	rooms    map[string]*Room
}

func NewClient(hub *Hub, conn *websocket.Conn, userID, username string) *Client {
	return &Client{
		hub:      hub,
		conn:     conn,
		send:     make(chan []byte, 256),
		userID:   userID,
		username: username,
		rooms:    make(map[string]*Room),
	}
}

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 {
		_, message, err := c.conn.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
				log.Printf("read error: %v", err)
			}
			break
		}

		var msg Message
		if err := json.Unmarshal(message, &msg); err != nil {
			log.Printf("invalid message: %v", err)
			continue
		}

		msg.Sender = c.userID
		msg.Timestamp = time.Now().UnixMilli()
		c.hub.HandleMessage(c, &msg)
	}
}

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)

			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
			}
		}
	}
}

アーキテクチャ2：Roomチャットルームとメッセージブロードキャスト

// internal/chat/room.go
package chat

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

type Room struct {
	ID      string
	clients map[*Client]bool
	forward chan *Message
	join    chan *Client
	leave   chan *Client
	mu      sync.RWMutex
}

type Message struct {
	Type      string `json:"type"`
	RoomID    string `json:"roomId"`
	Sender    string `json:"sender"`
	Content   string `json:"content"`
	Timestamp int64  `json:"timestamp"`
}

func NewRoom(id string) *Room {
	return &Room{
		ID:      id,
		clients: make(map[*Client]bool),
		forward: make(chan *Message, 100),
		join:    make(chan *Client),
		leave:   make(chan *Client),
	}
}

func (r *Room) Run() {
	for {
		select {
		case client := <-r.join:
			r.mu.Lock()
			r.clients[client] = true
			r.mu.Unlock()
			r.broadcastSystemMsg(client.username + " joined the room")

		case client := <-r.leave:
			r.mu.Lock()
			if _, ok := r.clients[client]; ok {
				delete(r.clients, client)
			}
			r.mu.Unlock()
			r.broadcastSystemMsg(client.username + " left the room")

		case msg := <-r.forward:
			r.mu.RLock()
			data, _ := json.Marshal(msg)
			for client := range r.clients {
				select {
				case client.send <- data:
				default:
					r.mu.RUnlock()
					r.mu.Lock()
					delete(r.clients, client)
					close(client.send)
					r.mu.Unlock()
					r.mu.RLock()
				}
			}
			r.mu.RUnlock()
		}
	}
}

func (r *Room) Join(client *Client) {
	r.join <- client
	client.rooms[r.ID] = r
}

func (r *Room) Leave(client *Client) {
	r.leave <- client
	delete(client.rooms, r.ID)
}

func (r *Room) Broadcast(msg *Message) {
	r.forward <- msg
}

func (r *Room) broadcastSystemMsg(content string) {
	msg := &Message{
		Type:      "system",
		RoomID:    r.ID,
		Content:   content,
		Timestamp: time.Now().UnixMilli(),
	}
	r.forward <- msg
}

アーキテクチャ3：HTTPアップグレードとWebSocketサーバー

// internal/chat/server.go
package chat

import (
	"log"
	"net/http"

	"github.com/gorilla/websocket"
)

var upgrader = websocket.Upgrader{
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
	CheckOrigin: func(r *http.Request) bool { return true },
}

type Server struct {
	hub *Hub
}

func NewServer(hub *Hub) *Server {
	return &Server{hub: hub}
}

func (s *Server) ServeWS(w http.ResponseWriter, r *http.Request) {
	userID := r.URL.Query().Get("userId")
	username := r.URL.Query().Get("username")
	if userID == "" || username == "" {
		http.Error(w, "userId and username required", http.StatusBadRequest)
		return
	}

	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Printf("upgrade error: %v", err)
		return
	}

	client := NewClient(s.hub, conn, userID, username)
	s.hub.register <- client

	go client.WritePump()
	go client.ReadPump()
}

アーキテクチャ4：Redis Pub/Subクロスインスタンスメッセージ配信

// internal/chat/redis_pubsub.go
package chat

import (
	"context"
	"encoding/json"
	"log"

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

type RedisBroker struct {
	client  *redis.Client
	hub     *Hub
	channel string
}

func NewRedisBroker(addr, channel string, hub *Hub) *RedisBroker {
	rdb := redis.NewClient(&redis.Options{Addr: addr, PoolSize: 10})
	return &RedisBroker{client: rdb, hub: hub, channel: channel}
}

func (rb *RedisBroker) Start(ctx context.Context) error {
	sub := rb.client.Subscribe(ctx, rb.channel)
	if _, err := sub.Receive(ctx); err != nil {
		return err
	}

	ch := sub.Channel()
	go func() {
		for msg := range ch {
			var m Message
			if err := json.Unmarshal([]byte(msg.Payload), &m); err != nil {
				log.Printf("redis message parse error: %v", err)
				continue
			}
			if room, ok := rb.hub.rooms[m.RoomID]; ok {
				room.forward <- &m
			}
		}
	}()
	return nil
}

func (rb *RedisBroker) Publish(ctx context.Context, msg *Message) error {
	data, err := json.Marshal(msg)
	if err != nil {
		return err
	}
	return rb.client.Publish(ctx, rb.channel, data).Err()
}

アーキテクチャ5：メッセージ永続化と履歴検索

// internal/chat/repository.go
package chat

import (
	"context"
	"database/sql"
	"fmt"

	_ "github.com/lib/pq"
)

type MessageRepository struct {
	db *sql.DB
}

func NewMessageRepository(databaseURL string) (*MessageRepository, error) {
	db, err := sql.Open("postgres", databaseURL)
	if err != nil {
		return nil, fmt.Errorf("failed to open database: %w", err)
	}
	return &MessageRepository{db: db}, nil
}

func (r *MessageRepository) Save(ctx context.Context, msg *Message) error {
	_, err := r.db.ExecContext(ctx,
		`INSERT INTO chat_messages (room_id, sender, content, type, timestamp) VALUES ($1, $2, $3, $4, $5)`,
		msg.RoomID, msg.Sender, msg.Content, msg.Type, msg.Timestamp)
	return err
}

func (r *MessageRepository) GetHistory(ctx context.Context, roomID string, limit, offset int) ([]*Message, error) {
	rows, err := r.db.QueryContext(ctx,
		`SELECT room_id, sender, content, type, timestamp FROM chat_messages WHERE room_id = $1 ORDER BY timestamp DESC LIMIT $2 OFFSET $3`,
		roomID, limit, offset)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var messages []*Message
	for rows.Next() {
		msg := &Message{}
		if err := rows.Scan(&msg.RoomID, &msg.Sender, &msg.Content, &msg.Type, &msg.Timestamp); err != nil {
			return nil, err
		}
		messages = append(messages, msg)
	}
	return messages, nil
}

---

よくある落とし穴

落とし穴1：sendチャネルバッファオーバーフロー

// ❌ 間違い：バッファなしチャネル、遅いクライアントがブロードキャストをブロック
send chan []byte

// ✅ 正しい：バッファ付きチャネル + オーバーフロー時切断
send chan []byte  // 256容量のバッファ付きチャネル
select {
case client.send <- data:
default:
    delete(r.clients, client)
    close(client.send)
}

落とし穴2：ハートビート検出の欠落

// ❌ 間違い：ping/pong未設定、切断を検出できない
c.conn.SetReadDeadline(time.Time{})

// ✅ 正しい：ping/pongハートビートを設定
c.conn.SetReadDeadline(time.Now().Add(pongWait))
c.conn.SetPongHandler(func(string) error {
    c.conn.SetReadDeadline(time.Now().Add(pongWait))
    return nil
})

落とし穴3：WebSocketへの並行書き込み

// ❌ 間違い：複数のgoroutineが同時に書き込み
go func() { c.conn.WriteMessage(1, data) }()
go func() { c.conn.WriteMessage(1, data2) }()

// ✅ 正しい：チャネル経由で書き込みを直列化
c.send <- data

落とし穴4：Closeメッセージの未処理

// ❌ 間違い：クローズフレームを無視
_, message, err := c.conn.ReadMessage()

// ✅ 正しい：正常終了と異常終了を区別
if err != nil {
    if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
        log.Printf("abnormal close: %v", err)
    }
    break
}

落とし穴5：Redis Pub/Subメッセージ損失

// ❌ 間違い：ブロードキャスト後に永続化、Redis切断時にメッセージ損失
room.Broadcast(msg)
redisBroker.Publish(ctx, msg)

// ✅ 正しい：永続化してからブロードキャスト
msgRepo.Save(ctx, msg)
room.Broadcast(msg)
redisBroker.Publish(ctx, msg)

---

エラートラブルシューティング

#	エラーメッセージ	原因	解決方法
1	websocket: close sent	接続が既に閉じている	クライアントの切断を確認
2	write: broken pipe	閉じた接続への書き込み	WritePumpでclose channelを処理
3	read: connection reset	クライアントの異常切断	ReadPumpでdeferクリーンアップ
4	upgrade: Origin check failed	CORSチェック失敗	upgrader.CheckOriginを設定
5	concurrent write to websocket	並行書き込み	チャネルで直列化
6	channel full	sendバッファ満杯	バッファ増加または遅いクライアントを切断
7	redis: connection refused	Redis未起動	Redis接続設定を確認
8	OOM killed	メモリオーバーフロー	接続数とメッセージバッファを制限
9	too many open files	ファイルディスクリプタ枯渇	ulimit -nを増加
10	ping timeout	ハートビートタイムアウト	pongWait時間を調整

---

高度な最適化

1. 接続レート制限：トークンバケットでIPごとの接続レートを制限
2. メッセージ圧縮：permessage-deflate拡張でWebSocketフレームを圧縮
3. シャードブロードキャスト：大規模ルームを接続シャードに分割してブロードキャスト
4. メッセージACK：確認応答機構で少なくとも1回の配信を保証
5. 接続マイグレーション：Redisで接続マッピングを記録、サーバー再起動後の再接続をサポート

---

比較分析

次元	gorilla/websocket	nhooyr.io/websocket	gobwas/ws
API使いやすさ	⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐
パフォーマンス	⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐⭐⭐
メモリ使用量	⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐⭐⭐
Contextサポート	⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐
圧縮サポート	⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐⭐
コミュニティ活発度	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐
ドキュメント品質	⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐

---

まとめ：Goのgoroutineモデルは高同時WebSocketサービス構築の天然の強みです。Hub-Client接続管理→Roomメッセージブロードキャスト→Redis Pub/Subクロスインスタンス→メッセージ永続化→水平スケーリングの5つが一体となり、ゼロから10万+同時接続のリアルタイムチャットシステムを構築できます。要点：書き込みの直列化、ハートビート検出、背圧制御、永続化してからブロードキャスト。

---

オンラインツール推奨

• JSONフォーマッター：/ja/json/format
• Hash計算：/ja/encode/hash
• cURL→コード変換：/ja/dev/curl-to-code