문서가격블로그
문의하기
로그인무료로 가입하기
Back to Tech Articles

数千の同時接続を処理するスケーラブルなGo WebSocketサービスの構築

7월 08, 2025
# Misc

Daniel Hayes

Full-Stack Engineer · Leapcell

이 게시물은 English, 한국어 및 Deutsch에서도 볼 수 있습니다.
Cover of "数千の同時接続を処理するスケーラブルなGo WebSocketサービスの構築"

はじめに

今日の相互接続された世界では、リアルタイム通信はもはや贅沢ではなく、期待となっています。ライブチャットアプリケーションや共同編集ツールからオンラインゲームや金融ダッシュボードまで、即時更新とインタラクティブなエクスペリエンスへの需要は増え続けています。クライアントとサーバー間の永続的な全二重通信チャネルを提供するWebSocketsは、そのようなアプリケーションを構築するための事実上の標準となっています。しかし、数千、さらには数百万もの同時WebSocket接続を処理することは、重大なエンジニアリング上の課題をもたらします。従来の要求-応答アーキテクチャは、この種の負荷の下では苦戦し、リソースの枯渇やパフォーマンスのボトルネックにつながることがよくあります。Goは、軽量なゴルーチンと効率的な並行モデルにより、高性能ネットワークサービスの構築に驚くほど適しています。この記事では、Goの強みを活用して、数千の同時接続を効果的に管理できるスケーラブルなWebSocketサーバーを構築する方法を探ります。これにより、堅牢なリアルタイムアプリケーションの基盤が築かれます。

コアコンポーネントの理解

実装の詳細に飛び込む前に、スケーラブルなWebSocketサービスを構築するために不可欠ないくつかの基本的な概念を明確にしましょう。

WebSockets

WebSocketsは、単一のTCP接続を介して永続的な双方向通信チャネルを提供します。ステートレスで要求-応答モデルに依存するHTTPとは異なり、WebSocketsは、初期ハンドシェイク後にクライアントとサーバーの両方がいつでもメッセージを送信できるようにすることで、オーバーヘッドとレイテンシを大幅に削減します。Goでは、github.com/gorilla/websocketライブラリは、WebSocketsを扱う上で最も人気のある選択肢であり、堅牢で使いやすいAPIを提供します。

Goroutines

Goroutinesは、Goの軽量で並行実行される関数です。従来のOSスレッドよりもはるかに安価であるため、Goプログラムは数千、さらには数百万ものゴルーチンを並行して起動できます。これは、多数のWebSocket接続を処理する際に重要な利点です。各接続は、大幅なリソースオーバーヘッドなしに独自のゴルーチンで管理できます。

Channels

Channelsは、ゴルーチンが値を送受信できる型付きのパイプです。ゴルーチン間の通信のために設計されており、競合状態を防ぐ安全なデータ共有メカニズムとして機能します。ChannelsはGoの並行モデルの基本であり、メッセージフローの管理とWebSocketサーバーでのゴルーチンのオーケストレーションに広く使用されます。

Fan-out/Fan-in パターン

これはGoで一般的な並行パターンです。「fan-out」フェーズは作業を複数のゴルーチンに分散し、「fan-in」フェーズはそれらのゴルーチンから結果を収集します。WebSocketのコンテキストでは、1つのクライアントからの単一のメッセージが複数のサブスクライブされたクライアントに「fan-out」される必要があり、さまざまなクライアントからのメッセージを中央処理ユニットに「fan-in」する場合があります。

スケーラブルなWebSocketサービスの構築

GoでスケーラブルなWebSocketサービスを構築するには、主に効率的な接続管理、メッセージブロードキャスト、リソース処理に焦点を当てた、いくつかの主要な設計上の考慮事項が必要です。

接続管理

各受信WebSocket接続は、受け入れて管理する必要があります。一般的なアプローチは、接続された各クライアントに1つのゴルーチンを割り当てることです。このゴルーチンは、クライアントからのメッセージの読み取り、クライアントへのメッセージの書き込み、および接続固有のロジックの処理を担当します。

package main

import (
	"log"
	"net/http"
	"time"

	"github.com/gorilla/websocket"
)

// Upgrader upgrades HTTP connections to WebSocket connections.
var upgrader = websocket.Upgrader{
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
	CheckOrigin: func(r *http.Request) bool {
		// Allow all origins for simplicity in this example.
		// In production, restrict this to your domain.
		return true
	},
}

// Client represents a single connected WebSocket client.
type Client struct {
	conn *websocket.Conn
	send chan []byte // Channel to send messages to the client
}

// readPump reads messages from the WebSocket connection.
func (c *Client) readPump() {
	defer func() {
		// Clean up the client connection when the goroutine exits
		log.Printf("Client disconnected: %s", c.conn.RemoteAddr())
		// TODO: Unregister client from the hub
		c.conn.Close()
	}()

	for {
		_, message, err := c.conn.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
				log.Printf("Read error: %v", err)
			}
			break
		}
		log.Printf("Received: %s", message)
		// TODO: Process received message (e.g., broadcast to others)
	}
}

// writePump writes messages to the WebSocket connection.
func (c *Client) writePump() {
	ticker := time.NewTicker(time.Second * 10) // Ping interval
	defer func() {
		ticker.Stop()
		c.conn.Close()
	}()

	for {
		select {
		case message, ok := <-c.send:
			if !ok {
				// The hub closed the channel.
				c.conn.WriteMessage(websocket.CloseMessage, []byte{})
				return
			}
			if err := c.conn.WriteMessage(websocket.TextMessage, message); err != nil {
				log.Printf("Write error: %v", err)
				return
			}
		case <-ticker.C:
			// Send ping messages to keep the connection alive
			if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
				log.Printf("Ping error: %v", err)
				return
			}
		}
	}
}

// serveWs handles WebSocket requests from peers.
func serveWs(w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Println("Upgrade error:", err)
		return
	}

	client := &Client{conn: conn, send: make(chan []byte, 256)}
	log.Printf("Client connected: %s", conn.RemoteAddr())
	// TODO: Register client with the hub

	go client.writePump()
	client.readPump() // Blocks until client disconnects or error
}

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

serveWs関数では、WebSocketのアップグレードが成功した後、接続とバッファ付きチャネル（send）を保持する Client 構造体を作成します。この send チャネルは、メッセージプロデューサーとコンシューマーを分離し、デッドロックを防ぎ、バックプレッシャーを提供するのに不可欠です。readPumpゴルーチンはクライアントからメッセージを継続的に読み取り、writePumpゴルーチンは send チャネルからクライアントにメッセージを送信し、接続を維持するための定期的なpingメッセージも処理します。

メッセージブロードキャストのための集中ハブ

複数のクライアントへのメッセージブロードキャストを効率的に処理するには、集中化された「ハブ」が不可欠です。このハブは、すべてのアクティブなクライアント接続を管理し、メッセージ配信を容易にします。

package main

import (
	"log"
	"net/http"
	"time"

	"github.com/gorilla/websocket"
)

// ... (Client, upgrader, readPump, writePump definitions as above) ...

// Hub maintains the set of active clients and broadcasts messages to them.
type Hub struct {
	// Registered clients.
	clients map[*Client]bool

	// Inbound messages from the clients.
	broadcast chan []byte

	// Register requests from the clients.
	register chan *Client

	// Unregister requests from clients.
	unregister chan *Client
}

// NewHub creates and returns a new Hub instance.
func NewHub() *Hub {
	return &Hub{
		broadcast:  make(chan []byte),
		register:   make(chan *Client),
		unregister: make(chan *Client),
		clients:    make(map[*Client]bool),
	}
}

// run starts the hub's main event loop.
func (h *Hub) run() {
	for {
		select {
		case client := <-h.register:
			h.clients[client] = true
			log.Printf("Client registered: %s (Total: %d)", client.conn.RemoteAddr(), len(h.clients))
		case client := <-h.unregister:
			if _, ok := h.clients[client]; ok {
				delete(h.clients, client)
				close(client.send) // Close the client's send channel
				log.Printf("Client unregistered: %s (Total: %d)", client.conn.RemoteAddr(), len(h.clients))
			}
		case message := <-h.broadcast:
			for client := range h.clients {
				select {
				case client.send <- message:
					// Message sent successfully
				default:
					// If send channel is full, assume client is slow or dead.
					// Unregister and close connection.
					close(client.send)
					delete(h.clients, client)
					log.Printf("Client send channel full, unregistering: %s", client.conn.RemoteAddr())
				}
			}
		}
	}
}

// serveWs handles WebSocket requests for connections.
func serveWs(hub *Hub, w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Println("Upgrade error:", err)
		return
	}
	client := &Client{conn: conn, send: make(chan []byte, 256)}
	hub.register <- client // Register the new client

	go client.writePump() // Client's write goroutine
	client.readPump()     // Client's read goroutine (blocks)

	// When readPump exits, unregister the client
	hub.unregister <- client
}

func main() {
	hub := NewHub()
	go hub.run() // Start the hub's goroutine

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

	// Example: Broadcast a message every 5 seconds
	go func() {
		for {
			time.Sleep(5 * time.Second)
			message := []byte("Hello from server!")
			select {
			case hub.broadcast <- message:
				log.Println("Broadcasting message:", string(message))
			default:
				log.Println("Hub broadcast channel full, skipping message.")
			}
		}
	}()

	log.Fatal(http.ListenAndServe(":8080", nil))
}

Hub構造体には3つのチャネル（register、unregister、broadcast）が含まれています。別のゴルーチン内の run メソッドは、これらのチャネルを継続的にリッスンします。

  • registerが新しいクライアントを受信すると、クライアントはその clients マップに追加されます。
  • unregisterがクライアントを受信すると、クライアントが削除され、その send チャネルが閉じられます。
  • broadcastがメッセージを受信すると、登録されているすべてのクライアントを反復処理し、各クライアントの send チャネルにメッセージを送信しようとします。クライアントの send チャネルがいっぱいの場合にブロッキングを防ぐために、default ケースを持つ select ステートメントが使用されます。これにより、低速なコンシューマーが他のクライアントに影響を与えることを防ぎます。

スケールへの最適化

数千の接続に対するスケーラビリティをさらに向上させるには：

  1. **バッファ付きチャネル：**クライアントの send キューには、十分にバッファリングされたチャネル（例： make(chan []byte, 256)）を使用します。これにより、クライアントが一時的に読み取りが遅い場合でも、サーバーがメッセージを送信できるようになり、バッファーが提供されます。
  2. **効率的なメッセージエンコーディング：**高スループットシナリオでは、JSONではなくProtocol BuffersやFlatBuffersのような効率的なバイナリシリアライゼーション形式を検討します。これにより、メッセージサイズと解析オーバーヘッドを削減できます。
  3. **水平スケーリング：**非常に多数の接続（数万または数百万）の場合、ロードバランサーの後ろに複数のGo WebSocketサーバーに接続を分散することを検討します。別のメッセージキュー（例：Kafka、NATS、Redis PubSub）を使用して、これらの独立したWebSocketサーバー間でメッセージを同期できます。各サーバーは関連するトピックをサブスクライブし、分散システム全体にメッセージを効果的にファンアウトします。
  4. **リソース管理：**メモリとCPUの使用率を注意深く監視します。ゴルーチンは軽量ですが、数千の接続は依然としてメモリを消費します。サーバーインフラストラクチャが、すべての接続とその関連バッファーの合計メモリフットプリントを処理できることを確認してください。
  5. **正常なシャットダウン：**サーバーが正常にシャットダウンでき、すべてのアクティブなWebSocket接続を閉じ、リソースをクリーンアップできるように、適切なシグナル処理を実装します。

結論

Goの強力な並行プリミティブを採用することで、スケーラブルなGo WebSocketサービスを構築することは可能です。各接続にゴルーチンを割り当て、クライアント管理とメッセージブロードキャストのための集中ハブを採用し、効率的なメッセージパッシングのためのバッファ付きチャネルを活用することで、堅牢性と高性能で数千の同時WebSocket接続を処理できます。github.com/gorilla/websocketライブラリはしっかりした基盤を提供し、接続管理、メッセージフロー、リソース最適化に関する慎重な設計により、Goは洗練されたリアルタイムアプリケーションの構築に優れた選択肢として際立っています。鍵は、ゴルーチンとチャネルを効果的に使用して並行処理を管理し、アプリケーションが需要に合わせてスムーズにスケーリングできるようにする、回復力があり誤り耐性のあるアーキテクチャを設計することにあります。

Share this article

HackerNews

More Posts from Leapcell

Popular Posts