Posted inプログラミング

Server-Sent Events (SSE) と Go によるリアルタイムデータストリーミング：並行処理と高スループットパイプライン

Programming tutorial - IT technology blog
Programming tutorial - IT technology blog

リアルタイム通信：最適なプロトコルの選択

ライブ通知フィード、オーダーブック（板情報）のティッカー、インフラのダッシュボードには共通の要件があります。それは、ブラウザへデータを即座にプッシュ配信することです。標準的なHTTPポーリングではこの要件に対応しきれません。10,000台のクライアントが毎秒サーバーに問い合わせを行うと、毎秒10,000件のリクエストに対してHTTPヘッダーの解析、TLSオーバーヘッド、データベースクエリの負荷が発生してしまいます。

現代のアーキテクチャにおいて、リアルタイム更新の実現方法は主に以下の3つに絞られます。

  • ショート／ロングポーリング: クライアントが繰り返しポーリングを行うか、新しいデータが届くまで接続を開いたまま維持します。特別なインフラなしで動作しますが、ポーリングのたびにヘッダーで帯域を浪費し（通常リクエストあたり500B〜1KB）、接続スロットを圧迫します。
  • WebSocket: 標準的なHTTPを永続的な双方向TCP接続へとアップグレードします。マルチプレイヤーゲームやリアルタイム共同編集ホワイトボードのような全二重通信が必要なワークロードにおける標準技術です。ただし、独自の再接続ロジックやハートビート、専用のプロキシ設定が必要になります。
  • Server-Sent Events (SSE): HTTP上に直接構築された軽量な標準規格です。サーバーは永続的なレスポンスを開いたまま維持し、クライアントへプレーンなUTF-8テキストをストリーミング配信します。ブラウザによる自動再接続を標準でサポートしており、エンタープライズプロキシでも問題なくルーティング可能で、HTTP/2の多重化もそのまま活用できます。

Server-Sent Events を採用すべきケース

SSEは、クライアントが同じソケット経由でデータを送信する必要がなく、データを受信するだけでよい単方向フィードに最適な選択肢です。

主なメリット

  • ネットワークとの高い互換性: SSEは標準のポート80および443でストリーミングを行います。ロードバランサー、APIゲートウェイ、企業内ファイアウォールでも、特別なWebSocketアップグレードハンドシェイクを必要とせず、通常のHTTPトラフィックとして扱われます。
  • 標準で備わった耐障害性: ブラウザネイティブのEventSource APIは、接続が切断された場合に自動で再接続を行います。さらにLast-Event-IDヘッダーを送信するため、バックエンドは中断された箇所からストリームを再開できます。
  • 極めて小さなメモリフットプリント: Goはgoroutineごとに約2KBの初期スタックメモリしか割り当てません。4GBのRAMを搭載した軽量なGoサーバーであれば、負荷に苦しむことなく50,000のSSE同時接続を余裕で処理できます。

トレードオフと制約

  • 完全な単方向通信: データはサーバーからクライアントへの一方向のみに流れます。クライアントからのアップストリームメッセージには、個別のHTTP POSTまたはPUTリクエストが依然として必要です。
  • HTTP/1.1におけるブラウザの接続数制限: ブラウザはドメインごとに最大6つの同時HTTP/1.1接続という厳格な制限を課しています。そのため、単一のTCP接続上で数百のストリームを多重化できるHTTP/2環境でSSEを提供することを推奨します。

アーキテクチャ：スレッドセーフなGoブローカーパターン

何千ものクライアントが同時にサブスクライブする場合、並行処理の制御が最大の課題となります。安易な実装では、サーバーがブロードキャストしている最中にクライアントが切断されると、競合状態（レースコンディション）が発生しがちです。

Goにおいて、バッファなしチャネルやクローズされたチャネルへの直接書き込みは、パニックやgoroutineのブロックを引き起こします。同様に、複数のgoroutineから生のクライアントマップに対する読み書きを行うと、ランタイムクラッシュが即座に発生します。

中央集権型のBrokerパターンを用いれば、この問題をスマートに解決できます。単一の専用ループがクライアントレジストリを管理し、専用チャネルを通じてサブスクライブ、サブスクライブ解除、ブロードキャストイベントを順次処理します。

ステップ別の実装手順

ここでは、Go標準ライブラリのみを使用して堅牢なSSEブローカーを構築する方法を説明します。

ステップ1：スレッドセーフなブローカーの実装

まず、ブローカーの状態とイベントループを定義し、クライアント登録とメッセージのファンアウトを安全に管理します。

package main

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

type Client chan string

type Broker struct {
	clients        map[Client]bool
	newClients     chan Client
	closingClients chan Client
	notifier       chan string
}

func NewBroker() *Broker {
	return &Broker{
		clients:        make(map[Client]bool),
		newClients:     make(chan Client),
		closingClients: make(chan Client),
		notifier:       make(chan string),
	}
}

func (b *Broker) Start() {
	for {
		select {
		case s := <-b.newClients:
			b.clients[s] = true
			log.Printf("クライアントが接続しました。アクティブなクライアント数: %d", len(b.clients))

		case s := <-b.closingClients:
			if _, ok := b.clients[s]; ok {
				delete(b.clients, s)
				close(s)
				log.Printf("クライアントが切断されました。アクティブなクライアント数: %d", len(b.clients))
			}

		case event := <-b.notifier:
			for s := range b.clients {
				select {
				case s <- event:
				default:
					// バッファがいっぱいかクライアントの処理が遅いため、ブローカー全体のブロックを回避
					log.Println("遅延しているクライアントをスキップします")
				}
			}
		}
	}
}

ステップ2：HTTP SSEハンドラーの作成

次に、HTTPハンドラーを作成します。Flusherのサポートを検証し、ストリーミング用のヘッダーを設定、クライアントチャネルを登録して、データが到着次第すぐにフラッシュします。

func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "ストリーミングはサポートされていません", http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	w.Header().Set("Access-Control-Allow-Origin", "*")

	messageChan := make(Client, 10)
	b.newClients <- messageChan

	defer func() {
		b.closingClients <- messageChan
	}()

	ctx := r.Context()

	for {
		select {
		case <-ctx.Done():
			return
		case msg, ok := <-messageChan:
			if !ok {
				return
			}
			// SSE標準フォーマット: data: <payload>\n\n
			fmt.Fprintf(w, "data: %s\n\n", msg)
			flusher.Flush()
		}
	}
}

ステップ3：ライブテレメトリの配信

ブローカーを起動し、2秒ごとにCPUメトリクスを配信するバックグラウンドワーカーを接続します。

func main() {
	broker := NewBroker()
	go broker.Start()

	// リアルタイムデータイベントをシミュレート
	go func() {
		for {
			time.Sleep(2 * time.Second)
			payload := fmt.Sprintf(`{"timestamp": %d, "metric": "cpu_load", "value": %.2f}`, 
				time.Now().Unix(), 
				float64(time.Now().UnixNano()%100)/10.0)
			broker.notifier <- payload
		}
	}()

	http.Handle("/events", broker)

	log.Println("SSEサーバーが :8080 で起動しました。エンドポイント: /events")
	if err := http.ListenAndServe(":8080", nil); err != nil {
		log.Fatalf("サーバーエラー: %v", err)
	}
}

ステップ4：ストリームのテスト

ターミナルからバッファリング無効化フラグ -N を付けた curl を使用して、生のイベントストリームをテストします。

curl -N -H "Accept: text/event-stream" http://localhost:8080/events

ブラウザでストリームを受信するには、わずか数行の標準JavaScriptコードで対応できます。

const eventSource = new EventSource('http://localhost:8080/events');

eventSource.onmessage = (event) => {
    const data = JSON.parse(event.data);
    console.log('リアルタイムメトリクスを受信:', data);
};

eventSource.onerror = (err) => {
    console.error('SSE接続エラー:', err);
};
Share:
Tags: