Node.js で WebSocket 通信を実装する

現代の Web アプリケーション開発において、リアルタイム通信は欠かせない技術となっています。チャットアプリケーション、ライブ通知、リアルタイムダッシュボード、オンラインゲームなど、ユーザーに即座に情報を届ける機能の需要が急速に高まっています。本記事では、Node.js を使った WebSocket 通信の実装方法について、基本的な概念から実践的なアプリケーション開発まで、段階的に詳しく解説いたします。初心者の方でも理解しやすいよう、豊富なコード例と実際のエラー対処法を交えながら、本格的なリアルタイム通信システムを構築していきましょう。
WebSocket 通信の重要性と HTTP 通信との違い
WebSocket 通信を学ぶ前に、従来の HTTP 通信との違いを理解することが重要です。これにより、WebSocket がどのような場面で威力を発揮するかが明確になります。
HTTP 通信の特徴と制約
従来の Web 通信の主役である HTTP プロトコルには、以下のような特徴があります。
# | 項目 | HTTP 通信 | 課題 |
---|---|---|---|
1 | 通信方向 | 単方向(クライアント → サーバー) | サーバーからの能動的な通信ができない |
2 | 接続方式 | リクエスト・レスポンス型 | 毎回接続の確立・切断が必要 |
3 | 状態管理 | ステートレス | セッション管理が複雑 |
4 | リアルタイム性 | ポーリング方式 | 遅延とサーバー負荷が大きい |
HTTP 通信でのリアルタイム性の課題
typescript// 従来のHTTPポーリング方式の例
class HttpPollingClient {
private intervalId: NodeJS.Timer | null = null;
private apiUrl: string;
constructor(apiUrl: string) {
this.apiUrl = apiUrl;
}
// 定期的にサーバーをポーリング
public startPolling(interval: number = 1000): void {
this.intervalId = setInterval(async () => {
try {
const response = await fetch(
`${this.apiUrl}/messages`
);
const data = await response.json();
// 新しいメッセージがあれば処理
if (data.messages.length > 0) {
this.handleNewMessages(data.messages);
}
} catch (error) {
console.error('ポーリングエラー:', error);
}
}, interval);
}
private handleNewMessages(messages: any[]): void {
console.log('新しいメッセージ:', messages);
// メッセージの表示処理
}
public stopPolling(): void {
if (this.intervalId) {
clearInterval(this.intervalId);
this.intervalId = null;
}
}
}
上記のポーリング方式では、以下のような問題が発生します:
- 無駄なリクエスト: 更新がなくても定期的にリクエストを送信
- 遅延の発生: ポーリング間隔分の遅延が避けられない
- サーバー負荷: 多数のクライアントが同時にポーリングする負荷
WebSocket 通信の革新的な特徴
WebSocket プロトコルは、これらの課題を根本的に解決します。
# | 項目 | WebSocket 通信 | メリット |
---|---|---|---|
1 | 通信方向 | 双方向(クライアント ⇄ サーバー) | サーバーからの即座な通知が可能 |
2 | 接続方式 | 持続的接続 | 一度確立すれば継続的に使用可能 |
3 | 状態管理 | ステートフル | 接続状態を維持できる |
4 | リアルタイム性 | イベント駆動 | 即座にデータを送受信可能 |
WebSocket 通信の仕組み
typescript// WebSocketの基本的な通信フロー例
class WebSocketClient {
private ws: WebSocket | null = null;
private url: string;
constructor(url: string) {
this.url = url;
}
// WebSocket接続の確立
public connect(): Promise<void> {
return new Promise((resolve, reject) => {
this.ws = new WebSocket(this.url);
// 接続成功時
this.ws.onopen = () => {
console.log('WebSocket接続が確立されました');
resolve();
};
// メッセージ受信時(サーバーからのリアルタイム通知)
this.ws.onmessage = (event) => {
const data = JSON.parse(event.data);
console.log('サーバーからのメッセージ:', data);
this.handleMessage(data);
};
// エラー発生時
this.ws.onerror = (error) => {
console.error('WebSocketエラー:', error);
reject(error);
};
// 接続終了時
this.ws.onclose = () => {
console.log('WebSocket接続が終了しました');
this.handleDisconnect();
};
});
}
// メッセージ送信(即座にサーバーに届く)
public sendMessage(message: any): void {
if (this.ws && this.ws.readyState === WebSocket.OPEN) {
this.ws.send(JSON.stringify(message));
}
}
private handleMessage(data: any): void {
// リアルタイムメッセージの処理
// UIの即座な更新などを行う
}
private handleDisconnect(): void {
// 再接続処理などを実装
}
}
実際の活用シーンと選択基準
WebSocket と HTTP の使い分けは、アプリケーションの要件によって決まります。
typescript// 通信方式の選択ガイド
class CommunicationSelector {
public static selectProtocol(requirements: {
realTimeRequired: boolean;
bidirectional: boolean;
connectionPersistence: boolean;
dataFrequency: 'low' | 'medium' | 'high';
}): 'http' | 'websocket' {
// リアルタイム性が重要な場合
if (requirements.realTimeRequired) {
return 'websocket';
}
// 双方向通信が必要な場合
if (requirements.bidirectional) {
return 'websocket';
}
// 高頻度なデータ通信の場合
if (requirements.dataFrequency === 'high') {
return 'websocket';
}
// 接続の持続が必要な場合
if (requirements.connectionPersistence) {
return 'websocket';
}
// その他の一般的な用途
return 'http';
}
}
// 使用例
const chatAppRequirements = {
realTimeRequired: true,
bidirectional: true,
connectionPersistence: true,
dataFrequency: 'medium' as const,
};
console.log(
CommunicationSelector.selectProtocol(chatAppRequirements)
); // 'websocket'
Node.js での WebSocket 実装の基本概念
Node.js で WebSocket 通信を実装する際に理解しておくべき基本概念を詳しく説明します。
WebSocket ライブラリの比較と選択
Node.js エコシステムには、複数の WebSocket ライブラリが存在します。
# | ライブラリ | 特徴 | 適用場面 | 学習コスト |
---|---|---|---|---|
1 | ws | 軽量・高速・標準準拠 | シンプルな WebSocket 通信 | 低 |
2 | Socket.IO | 高機能・フォールバック対応 | 複雑なリアルタイムアプリ | 中 |
3 | uWS | 超高速・C++実装 | 高負荷・パフォーマンス重視 | 高 |
4 | ws + Express | Express 統合 | 既存の Express アプリ拡張 | 低 |
基本的な依存関係の設定
まず、プロジェクトのセットアップから始めましょう。
bash# プロジェクトの初期化
yarn init -y
# 基本的な依存関係をインストール
yarn add ws express
yarn add -D @types/ws @types/express typescript ts-node nodemon
# TypeScript設定ファイルの作成
npx tsc --init
json// package.json の scripts セクション
{
"scripts": {
"dev": "nodemon --exec ts-node src/server.ts",
"build": "tsc",
"start": "node dist/server.js"
}
}
Node.js のイベントループと WebSocket
WebSocket は非同期処理であるため、Node.js のイベントループの理解が重要です。
typescriptimport { EventEmitter } from 'events';
import WebSocket from 'ws';
// WebSocketサーバーのイベント管理クラス
class WebSocketEventManager extends EventEmitter {
private clients: Set<WebSocket> = new Set();
// クライアント接続の管理
public addClient(ws: WebSocket): void {
this.clients.add(ws);
console.log(`クライアント接続数: ${this.clients.size}`);
// クライアント切断時の処理
ws.on('close', () => {
this.clients.delete(ws);
console.log(
`クライアント切断: 接続数 ${this.clients.size}`
);
this.emit('clientDisconnected', ws);
});
// メッセージ受信時の処理
ws.on('message', (data) => {
try {
const message = JSON.parse(data.toString());
this.emit('messageReceived', ws, message);
} catch (error) {
console.error('メッセージパースエラー:', error);
this.emit('parseError', ws, error);
}
});
this.emit('clientConnected', ws);
}
// 全クライアントへのブロードキャスト
public broadcast(message: any): void {
const messageString = JSON.stringify(message);
this.clients.forEach((client) => {
if (client.readyState === WebSocket.OPEN) {
client.send(messageString);
}
});
}
// 特定クライアントへのメッセージ送信
public sendToClient(ws: WebSocket, message: any): void {
if (ws.readyState === WebSocket.OPEN) {
ws.send(JSON.stringify(message));
}
}
// 接続中のクライアント数を取得
public getClientCount(): number {
return this.clients.size;
}
}
ws ライブラリを使った基本的な WebSocket サーバーの構築
実際に WebSocket サーバーを構築していきましょう。段階的に機能を追加していきます。
最小限の WebSocket サーバー
typescript// src/basic-websocket-server.ts
import WebSocket, { WebSocketServer } from 'ws';
import http from 'http';
// HTTPサーバーの作成
const server = http.createServer();
const wss = new WebSocketServer({ server });
// WebSocket接続イベント
wss.on('connection', (ws: WebSocket) => {
console.log('新しいクライアントが接続しました');
// エコーサーバーの実装(受信したメッセージをそのまま返す)
ws.on('message', (data: WebSocket.Data) => {
console.log('受信:', data.toString());
// 受信したメッセージをクライアントに送り返す
ws.send(`エコー: ${data.toString()}`);
});
// 接続終了時の処理
ws.on('close', () => {
console.log('クライアントが切断しました');
});
// エラー処理
ws.on('error', (error) => {
console.error('WebSocketエラー:', error);
});
// 接続確立の確認メッセージ
ws.send('WebSocketサーバーに接続されました');
});
// サーバー起動
const PORT = process.env.PORT || 8080;
server.listen(PORT, () => {
console.log(
`WebSocketサーバーがポート${PORT}で起動しました`
);
});
高機能な WebSocket サーバーの実装
typescript// src/advanced-websocket-server.ts
import WebSocket, { WebSocketServer } from 'ws';
import express from 'express';
import http from 'http';
import path from 'path';
import { v4 as uuidv4 } from 'uuid';
// クライアント情報の型定義
interface ClientInfo {
id: string;
ws: WebSocket;
username?: string;
joinedAt: Date;
lastActivity: Date;
}
// メッセージの型定義
interface Message {
type: 'join' | 'message' | 'leave' | 'ping' | 'error';
clientId: string;
username?: string;
content?: string;
timestamp: Date;
}
class AdvancedWebSocketServer {
private app: express.Application;
private server: http.Server;
private wss: WebSocketServer;
private clients: Map<string, ClientInfo> = new Map();
constructor() {
this.app = express();
this.server = http.createServer(this.app);
this.wss = new WebSocketServer({ server: this.server });
this.setupExpressRoutes();
this.setupWebSocketHandlers();
this.setupHeartbeat();
}
// Express.jsルートの設定
private setupExpressRoutes(): void {
// 静的ファイルの提供
this.app.use(express.static('public'));
// API エンドポイント:接続中のクライアント情報
this.app.get('/api/clients', (req, res) => {
const clientList = Array.from(
this.clients.values()
).map((client) => ({
id: client.id,
username: client.username,
joinedAt: client.joinedAt,
lastActivity: client.lastActivity,
}));
res.json({
totalClients: this.clients.size,
clients: clientList,
});
});
// ヘルスチェックエンドポイント
this.app.get('/health', (req, res) => {
res.json({
status: 'ok',
timestamp: new Date(),
connectedClients: this.clients.size,
uptime: process.uptime(),
});
});
}
// WebSocketイベントハンドラーの設定
private setupWebSocketHandlers(): void {
this.wss.on('connection', (ws: WebSocket, request) => {
const clientId = uuidv4();
const clientInfo: ClientInfo = {
id: clientId,
ws,
joinedAt: new Date(),
lastActivity: new Date(),
};
this.clients.set(clientId, clientInfo);
console.log(
`クライアント接続: ${clientId} (総接続数: ${this.clients.size})`
);
// 接続確認メッセージ
this.sendToClient(clientId, {
type: 'join',
clientId,
content: 'WebSocketサーバーに接続されました',
timestamp: new Date(),
});
// メッセージ受信処理
ws.on('message', (data: WebSocket.Data) => {
try {
const message: Message = JSON.parse(
data.toString()
);
this.handleMessage(clientId, message);
} catch (error) {
console.error(
`メッセージパースエラー [${clientId}]:`,
error
);
this.sendError(clientId, 'Invalid JSON format');
}
});
// 接続終了処理
ws.on('close', () => {
this.handleClientDisconnect(clientId);
});
// エラー処理
ws.on('error', (error) => {
console.error(
`WebSocketエラー [${clientId}]:`,
error
);
this.handleClientDisconnect(clientId);
});
});
}
// メッセージ処理
private handleMessage(
clientId: string,
message: Message
): void {
const client = this.clients.get(clientId);
if (!client) return;
// 最終活動時刻を更新
client.lastActivity = new Date();
switch (message.type) {
case 'join':
this.handleJoinMessage(clientId, message);
break;
case 'message':
this.handleChatMessage(clientId, message);
break;
case 'ping':
this.handlePingMessage(clientId);
break;
default:
this.sendError(
clientId,
`Unknown message type: ${message.type}`
);
}
}
// 参加メッセージの処理
private handleJoinMessage(
clientId: string,
message: Message
): void {
const client = this.clients.get(clientId);
if (!client) return;
client.username = message.username;
// 参加通知をブロードキャスト
this.broadcast(
{
type: 'join',
clientId,
username: message.username,
content: `${message.username}さんが参加しました`,
timestamp: new Date(),
},
clientId
);
console.log(
`ユーザー参加: ${message.username} [${clientId}]`
);
}
// チャットメッセージの処理
private handleChatMessage(
clientId: string,
message: Message
): void {
const client = this.clients.get(clientId);
if (!client) return;
const chatMessage: Message = {
type: 'message',
clientId,
username: client.username,
content: message.content,
timestamp: new Date(),
};
// 全クライアントにメッセージをブロードキャスト
this.broadcast(chatMessage);
console.log(
`チャットメッセージ [${client.username}]: ${message.content}`
);
}
// Pingメッセージの処理(ハートビート)
private handlePingMessage(clientId: string): void {
this.sendToClient(clientId, {
type: 'ping',
clientId,
content: 'pong',
timestamp: new Date(),
});
}
// 特定クライアントへのメッセージ送信
private sendToClient(
clientId: string,
message: Message
): void {
const client = this.clients.get(clientId);
if (client && client.ws.readyState === WebSocket.OPEN) {
client.ws.send(JSON.stringify(message));
}
}
// 全クライアントへのブロードキャスト
private broadcast(
message: Message,
excludeClientId?: string
): void {
const messageString = JSON.stringify(message);
this.clients.forEach((client, clientId) => {
if (
clientId !== excludeClientId &&
client.ws.readyState === WebSocket.OPEN
) {
client.ws.send(messageString);
}
});
}
// エラーメッセージの送信
private sendError(
clientId: string,
errorMessage: string
): void {
this.sendToClient(clientId, {
type: 'error',
clientId,
content: errorMessage,
timestamp: new Date(),
});
}
// クライアント切断処理
private handleClientDisconnect(clientId: string): void {
const client = this.clients.get(clientId);
if (!client) return;
// 離脱通知をブロードキャスト
if (client.username) {
this.broadcast(
{
type: 'leave',
clientId,
username: client.username,
content: `${client.username}さんが退出しました`,
timestamp: new Date(),
},
clientId
);
}
this.clients.delete(clientId);
console.log(
`クライアント切断: ${clientId} (総接続数: ${this.clients.size})`
);
}
// ハートビート機能(非アクティブなクライアントの検出)
private setupHeartbeat(): void {
setInterval(() => {
const now = new Date();
const timeout = 30000; // 30秒
this.clients.forEach((client, clientId) => {
const timeSinceLastActivity =
now.getTime() - client.lastActivity.getTime();
if (timeSinceLastActivity > timeout) {
console.log(
`非アクティブクライアントを切断: ${clientId}`
);
client.ws.terminate();
this.handleClientDisconnect(clientId);
}
});
}, 10000); // 10秒ごとにチェック
}
// サーバー起動
public start(port: number = 8080): void {
this.server.listen(port, () => {
console.log(
`高機能WebSocketサーバーがポート${port}で起動しました`
);
console.log(
`HTTP API: http://localhost:${port}/api/clients`
);
console.log(
`ヘルスチェック: http://localhost:${port}/health`
);
});
}
}
// サーバーのインスタンス化と起動
const wsServer = new AdvancedWebSocketServer();
wsServer.start(8080);
この基本的な WebSocket サーバーの実装では、以下の機能を提供しています:
- 基本的な WebSocket 通信: メッセージの送受信
- Express.js 統合: HTTP エンドポイントとの併用
- クライアント管理: 接続中のクライアント情報の追跡
- ブロードキャスト機能: 全クライアントへのメッセージ配信
- ハートビート: 非アクティブなクライアントの自動切断
- エラーハンドリング: 適切なエラー処理と通知
次のセクションでは、クライアントサイドの実装について詳しく説明いたします。
クライアントサイドでの WebSocket 接続実装
サーバーサイドの実装が完了したので、次はクライアントサイドの WebSocket 接続を実装していきましょう。
ブラウザ向け WebSocket クライアントの実装
typescript// public/js/websocket-client.ts
interface Message {
type: 'join' | 'message' | 'leave' | 'ping' | 'error';
clientId: string;
username?: string;
content?: string;
timestamp: Date;
}
class WebSocketClient {
private ws: WebSocket | null = null;
private url: string;
private reconnectAttempts: number = 0;
private maxReconnectAttempts: number = 5;
private reconnectInterval: number = 1000;
private heartbeatInterval: NodeJS.Timer | null = null;
// イベントハンドラー
private onMessageHandler?: (message: Message) => void;
private onConnectHandler?: () => void;
private onDisconnectHandler?: () => void;
private onErrorHandler?: (error: Event) => void;
constructor(url: string) {
this.url = url;
}
// WebSocket接続の確立
public async connect(): Promise<void> {
return new Promise((resolve, reject) => {
try {
this.ws = new WebSocket(this.url);
// 接続成功時
this.ws.onopen = () => {
console.log('WebSocket接続が確立されました');
this.reconnectAttempts = 0;
this.startHeartbeat();
if (this.onConnectHandler) {
this.onConnectHandler();
}
resolve();
};
// メッセージ受信時
this.ws.onmessage = (event) => {
try {
const message: Message = JSON.parse(event.data);
console.log('受信メッセージ:', message);
if (this.onMessageHandler) {
this.onMessageHandler(message);
}
} catch (error) {
console.error('メッセージパースエラー:', error);
}
};
// 接続終了時
this.ws.onclose = () => {
console.log('WebSocket接続が終了しました');
this.stopHeartbeat();
if (this.onDisconnectHandler) {
this.onDisconnectHandler();
}
// 自動再接続の試行
this.attemptReconnect();
};
// エラー発生時
this.ws.onerror = (error) => {
console.error('WebSocketエラー:', error);
if (this.onErrorHandler) {
this.onErrorHandler(error);
}
reject(error);
};
} catch (error) {
console.error(
'WebSocket接続の初期化に失敗:',
error
);
reject(error);
}
});
}
// メッセージ送信
public sendMessage(message: Partial<Message>): void {
if (this.ws && this.ws.readyState === WebSocket.OPEN) {
const fullMessage: Message = {
type: message.type || 'message',
clientId: message.clientId || '',
username: message.username,
content: message.content,
timestamp: new Date(),
};
this.ws.send(JSON.stringify(fullMessage));
} else {
console.error('WebSocket接続が確立されていません');
}
}
// チャットルームに参加
public joinChat(username: string): void {
this.sendMessage({
type: 'join',
username: username,
content: `${username}がチャットに参加しました`,
});
}
// チャットメッセージの送信
public sendChatMessage(content: string): void {
this.sendMessage({
type: 'message',
content: content,
});
}
// ハートビート(接続維持)
private startHeartbeat(): void {
this.heartbeatInterval = setInterval(() => {
this.sendMessage({ type: 'ping' });
}, 30000); // 30秒ごと
}
private stopHeartbeat(): void {
if (this.heartbeatInterval) {
clearInterval(this.heartbeatInterval);
this.heartbeatInterval = null;
}
}
// 自動再接続
private attemptReconnect(): void {
if (
this.reconnectAttempts < this.maxReconnectAttempts
) {
this.reconnectAttempts++;
console.log(
`再接続を試行中... (${this.reconnectAttempts}/${this.maxReconnectAttempts})`
);
setTimeout(() => {
this.connect().catch(() => {
console.log('再接続に失敗しました');
});
}, this.reconnectInterval * this.reconnectAttempts);
} else {
console.error('最大再接続試行回数に達しました');
}
}
// イベントハンドラーの設定
public onMessage(
handler: (message: Message) => void
): void {
this.onMessageHandler = handler;
}
public onConnect(handler: () => void): void {
this.onConnectHandler = handler;
}
public onDisconnect(handler: () => void): void {
this.onDisconnectHandler = handler;
}
public onError(handler: (error: Event) => void): void {
this.onErrorHandler = handler;
}
// 接続終了
public disconnect(): void {
if (this.ws) {
this.ws.close();
this.ws = null;
}
this.stopHeartbeat();
}
// 接続状態の確認
public isConnected(): boolean {
return (
this.ws !== null &&
this.ws.readyState === WebSocket.OPEN
);
}
}
React 統合の WebSocket クライアント
Next.js や React アプリケーションで WebSocket を使用する場合の実装例です。
typescript// hooks/useWebSocket.ts
import {
useEffect,
useRef,
useState,
useCallback,
} from 'react';
interface Message {
type: 'join' | 'message' | 'leave' | 'ping' | 'error';
clientId: string;
username?: string;
content?: string;
timestamp: Date;
}
interface UseWebSocketReturn {
messages: Message[];
connectionStatus:
| 'connecting'
| 'connected'
| 'disconnected'
| 'error';
sendMessage: (message: Partial<Message>) => void;
joinChat: (username: string) => void;
sendChatMessage: (content: string) => void;
isConnected: boolean;
}
export const useWebSocket = (
url: string
): UseWebSocketReturn => {
const [messages, setMessages] = useState<Message[]>([]);
const [connectionStatus, setConnectionStatus] = useState<
'connecting' | 'connected' | 'disconnected' | 'error'
>('disconnected');
const wsRef = useRef<WebSocket | null>(null);
const reconnectTimeoutRef = useRef<NodeJS.Timeout | null>(
null
);
const reconnectAttempts = useRef(0);
const maxReconnectAttempts = 5;
// WebSocket接続の確立
const connect = useCallback(() => {
try {
setConnectionStatus('connecting');
wsRef.current = new WebSocket(url);
wsRef.current.onopen = () => {
console.log('WebSocket接続が確立されました');
setConnectionStatus('connected');
reconnectAttempts.current = 0;
};
wsRef.current.onmessage = (event) => {
try {
const message: Message = JSON.parse(event.data);
// メッセージを状態に追加
setMessages((prev) => [...prev, message]);
} catch (error) {
console.error('メッセージパースエラー:', error);
}
};
wsRef.current.onclose = () => {
console.log('WebSocket接続が終了しました');
setConnectionStatus('disconnected');
// 自動再接続
if (
reconnectAttempts.current < maxReconnectAttempts
) {
reconnectAttempts.current++;
const delay = 1000 * reconnectAttempts.current;
console.log(
`${delay}ms後に再接続を試行します (${reconnectAttempts.current}/${maxReconnectAttempts})`
);
reconnectTimeoutRef.current = setTimeout(() => {
connect();
}, delay);
}
};
wsRef.current.onerror = (error) => {
console.error('WebSocketエラー:', error);
setConnectionStatus('error');
};
} catch (error) {
console.error('WebSocket接続の初期化に失敗:', error);
setConnectionStatus('error');
}
}, [url]);
// メッセージ送信
const sendMessage = useCallback(
(message: Partial<Message>) => {
if (
wsRef.current &&
wsRef.current.readyState === WebSocket.OPEN
) {
const fullMessage: Message = {
type: message.type || 'message',
clientId: message.clientId || '',
username: message.username,
content: message.content,
timestamp: new Date(),
};
wsRef.current.send(JSON.stringify(fullMessage));
} else {
console.error('WebSocket接続が確立されていません');
}
},
[]
);
// チャット参加
const joinChat = useCallback(
(username: string) => {
sendMessage({
type: 'join',
username: username,
});
},
[sendMessage]
);
// チャットメッセージ送信
const sendChatMessage = useCallback(
(content: string) => {
sendMessage({
type: 'message',
content: content,
});
},
[sendMessage]
);
// 初期接続とクリーンアップ
useEffect(() => {
connect();
return () => {
if (reconnectTimeoutRef.current) {
clearTimeout(reconnectTimeoutRef.current);
}
if (wsRef.current) {
wsRef.current.close();
}
};
}, [connect]);
return {
messages,
connectionStatus,
sendMessage,
joinChat,
sendChatMessage,
isConnected: connectionStatus === 'connected',
};
};
リアルタイムチャット機能の実装
WebSocket クライアントができたので、実際のチャットアプリケーションを構築してみましょう。
React ベースのチャットコンポーネント
typescript// components/ChatApp.tsx
import React, { useState, useRef, useEffect } from 'react';
import { useWebSocket } from '../hooks/useWebSocket';
interface ChatMessage {
id: string;
username: string;
content: string;
timestamp: Date;
type: 'message' | 'join' | 'leave';
}
const ChatApp: React.FC = () => {
const [username, setUsername] = useState<string>('');
const [currentMessage, setCurrentMessage] =
useState<string>('');
const [isJoined, setIsJoined] = useState<boolean>(false);
const [chatMessages, setChatMessages] = useState<
ChatMessage[]
>([]);
const messagesEndRef = useRef<HTMLDivElement>(null);
// WebSocketフックの使用
const {
messages,
connectionStatus,
joinChat,
sendChatMessage,
isConnected,
} = useWebSocket('ws://localhost:8080');
// メッセージをチャットメッセージ形式に変換
useEffect(() => {
const newChatMessages = messages.map((msg) => ({
id: `${msg.clientId}-${msg.timestamp}`,
username: msg.username || 'Anonymous',
content: msg.content || '',
timestamp: new Date(msg.timestamp),
type: msg.type as 'message' | 'join' | 'leave',
}));
setChatMessages(newChatMessages);
}, [messages]);
// 自動スクロール
useEffect(() => {
messagesEndRef.current?.scrollIntoView({
behavior: 'smooth',
});
}, [chatMessages]);
// チャット参加
const handleJoinChat = () => {
if (username.trim() && isConnected) {
joinChat(username.trim());
setIsJoined(true);
}
};
// メッセージ送信
const handleSendMessage = () => {
if (currentMessage.trim() && isConnected && isJoined) {
sendChatMessage(currentMessage.trim());
setCurrentMessage('');
}
};
// Enterキーでメッセージ送信
const handleKeyPress = (e: React.KeyboardEvent) => {
if (e.key === 'Enter' && !e.shiftKey) {
e.preventDefault();
handleSendMessage();
}
};
// 接続状態のインジケーター
const renderConnectionStatus = () => {
const statusConfig = {
connecting: {
color: 'text-yellow-500',
text: '接続中...',
},
connected: {
color: 'text-green-500',
text: '接続済み',
},
disconnected: {
color: 'text-red-500',
text: '切断済み',
},
error: { color: 'text-red-500', text: 'エラー' },
};
const config = statusConfig[connectionStatus];
return (
<div
className={`mb-4 p-2 rounded ${config.color} bg-gray-100`}
>
ステータス: {config.text}
</div>
);
};
// メッセージの表示
const renderMessage = (message: ChatMessage) => {
const isSystemMessage =
message.type === 'join' || message.type === 'leave';
return (
<div
key={message.id}
className={`mb-2 p-2 rounded ${
isSystemMessage
? 'bg-gray-200 text-gray-600 italic'
: 'bg-blue-50'
}`}
>
<div className='flex justify-between items-center'>
<span className='font-semibold'>
{message.username}
</span>
<span className='text-xs text-gray-500'>
{message.timestamp.toLocaleTimeString()}
</span>
</div>
<div className='mt-1'>{message.content}</div>
</div>
);
};
if (!isJoined) {
// 参加画面
return (
<div className='max-w-md mx-auto mt-8 p-6 bg-white rounded-lg shadow-lg'>
<h2 className='text-2xl font-bold mb-4 text-center'>
チャットに参加
</h2>
{renderConnectionStatus()}
<div className='mb-4'>
<label
htmlFor='username'
className='block text-sm font-medium text-gray-700 mb-2'
>
ユーザー名
</label>
<input
type='text'
id='username'
value={username}
onChange={(e) => setUsername(e.target.value)}
placeholder='ユーザー名を入力'
className='w-full px-3 py-2 border border-gray-300 rounded-md focus:outline-none focus:ring-2 focus:ring-blue-500'
onKeyPress={(e) =>
e.key === 'Enter' && handleJoinChat()
}
/>
</div>
<button
onClick={handleJoinChat}
disabled={!username.trim() || !isConnected}
className='w-full bg-blue-500 text-white py-2 px-4 rounded-md hover:bg-blue-600 disabled:bg-gray-400 disabled:cursor-not-allowed'
>
参加する
</button>
</div>
);
}
// メインチャット画面
return (
<div className='max-w-2xl mx-auto mt-8 p-6 bg-white rounded-lg shadow-lg'>
<h2 className='text-2xl font-bold mb-4'>
リアルタイムチャット
</h2>
{renderConnectionStatus()}
{/* メッセージエリア */}
<div className='h-96 overflow-y-auto border border-gray-300 rounded-md p-4 mb-4 bg-gray-50'>
{chatMessages.length === 0 ? (
<div className='text-center text-gray-500'>
メッセージがありません。最初のメッセージを送信してください!
</div>
) : (
chatMessages.map(renderMessage)
)}
<div ref={messagesEndRef} />
</div>
{/* メッセージ入力エリア */}
<div className='flex space-x-2'>
<input
type='text'
value={currentMessage}
onChange={(e) =>
setCurrentMessage(e.target.value)
}
onKeyPress={handleKeyPress}
placeholder='メッセージを入力...'
className='flex-1 px-3 py-2 border border-gray-300 rounded-md focus:outline-none focus:ring-2 focus:ring-blue-500'
disabled={!isConnected}
/>
<button
onClick={handleSendMessage}
disabled={!currentMessage.trim() || !isConnected}
className='bg-blue-500 text-white px-6 py-2 rounded-md hover:bg-blue-600 disabled:bg-gray-400 disabled:cursor-not-allowed'
>
送信
</button>
</div>
{/* 統計情報 */}
<div className='mt-4 text-sm text-gray-600'>
<div>ユーザー名: {username}</div>
<div>
メッセージ数:{' '}
{
chatMessages.filter((m) => m.type === 'message')
.length
}
</div>
</div>
</div>
);
};
export default ChatApp;
エラーハンドリングとトラブルシューティング
実際の運用で遭遇する一般的なエラーとその対処法をご紹介します。
typescript// utils/websocket-error-handler.ts
interface WebSocketError {
code: number;
reason: string;
wasClean: boolean;
}
class WebSocketErrorHandler {
// 一般的なWebSocketエラーの処理
public static handleError(
error: WebSocketError | Event
): void {
if ('code' in error) {
// CloseEventの場合
this.handleCloseError(error as WebSocketError);
} else {
// 一般的なエラーイベント
console.error('WebSocket汎用エラー:', error);
}
}
private static handleCloseError(
error: WebSocketError
): void {
const { code, reason, wasClean } = error;
switch (code) {
case 1000:
console.log('正常にWebSocket接続が終了しました');
break;
case 1001:
console.log(
'エンドポイントが離脱しました (Going Away)'
);
// サーバーがシャットダウンまたはブラウザがページを離れた
break;
case 1002:
console.error('プロトコルエラーが発生しました');
// WebSocketプロトコル違反
break;
case 1003:
console.error(
'サポートされていないデータタイプを受信しました'
);
// バイナリデータの処理問題など
break;
case 1005:
console.log(
'ステータスコードが提供されませんでした'
);
// No Status Rcvd
break;
case 1006:
console.error('接続が異常終了しました');
// ネットワーク問題やサーバークラッシュ
this.handleAbnormalClosure();
break;
case 1007:
console.error(
'無効なペイロードデータが送信されました'
);
// UTF-8データの問題など
break;
case 1008:
console.error(
'ポリシー違反により接続が終了しました'
);
// セキュリティポリシー違反
break;
case 1009:
console.error('メッセージサイズが大きすぎます');
// Maximum message size exceeded
break;
case 1010:
console.error(
'必要な拡張機能がサポートされていません'
);
// Extension negotiation failure
break;
case 1011:
console.error(
'サーバーで予期しないエラーが発生しました'
);
// Internal server error
break;
default:
console.error(
`未知のWebSocketエラー: コード${code}, 理由: ${reason}`
);
}
if (!wasClean) {
console.warn(
'WebSocket接続が正常に終了しませんでした'
);
}
}
// 異常終了時の処理
private static handleAbnormalClosure(): void {
console.log(
'異常終了を検出。再接続処理を開始します...'
);
// 再接続ロジックを実行
}
// ネットワークエラーの診断
public static diagnoseNetworkError(): void {
console.log('ネットワーク診断を実行中...');
// オンライン状態のチェック
if (!navigator.onLine) {
console.error('インターネット接続がありません');
return;
}
// サーバー到達性のテスト
fetch('/health')
.then((response) => {
if (response.ok) {
console.log('HTTPサーバーは到達可能です');
console.log(
'WebSocketサーバーの設定を確認してください'
);
} else {
console.error(
'サーバーがエラーを返しています:',
response.status
);
}
})
.catch((error) => {
console.error('サーバーに到達できません:', error);
});
}
}
// 使用例
export { WebSocketErrorHandler };
このクライアントサイドの実装では、以下の機能を提供しています:
- 自動再接続: 接続が切断された場合の自動復旧
- React 統合: React アプリケーションでの簡単な利用
- エラーハンドリング: 詳細なエラー診断と対処
- ハートビート: 接続維持のための Ping/Pong
- TypeScript 対応: 型安全な WebSocket 通信
次のセクションでは、Socket.IO を使ったより高機能な WebSocket 通信について説明いたします。
Socket.IO を活用した高機能 WebSocket 通信
Socket.IO は、WebSocket をベースにしながら、フォールバック機能や高度な機能を提供するライブラリです。
Socket.IO の導入とセットアップ
bash# Socket.IOの依存関係をインストール
yarn add socket.io socket.io-client
yarn add -D @types/socket.io @types/socket.io-client
Socket.IO サーバーの実装
typescript// src/socket-io-server.ts
import express from 'express';
import { createServer } from 'http';
import { Server } from 'socket.io';
import cors from 'cors';
interface User {
id: string;
username: string;
room: string;
joinedAt: Date;
}
interface ChatMessage {
id: string;
username: string;
content: string;
room: string;
timestamp: Date;
}
class SocketIOChatServer {
private app: express.Application;
private server: any;
private io: Server;
private users: Map<string, User> = new Map();
private rooms: Map<string, Set<string>> = new Map();
constructor() {
this.app = express();
this.server = createServer(this.app);
this.io = new Server(this.server, {
cors: {
origin: '*',
methods: ['GET', 'POST'],
},
// 接続オプション
pingTimeout: 60000,
pingInterval: 25000,
});
this.setupMiddleware();
this.setupSocketHandlers();
this.setupRoomManagement();
}
private setupMiddleware(): void {
this.app.use(cors());
this.app.use(express.json());
this.app.use(express.static('public'));
// REST API エンドポイント:接続中のクライアント情報
this.app.get('/api/rooms', (req, res) => {
const roomList = Array.from(this.rooms.entries()).map(
([room, users]) => ({
name: room,
userCount: users.size,
users: Array.from(users)
.map((userId) => {
const user = this.users.get(userId);
return user
? { id: user.id, username: user.username }
: null;
})
.filter(Boolean),
})
);
res.json(roomList);
});
this.app.get('/api/users', (req, res) => {
const userList = Array.from(this.users.values());
res.json({
total: userList.length,
users: userList,
});
});
}
private setupSocketHandlers(): void {
// 名前空間の設定
const chatNamespace = this.io.of('/chat');
chatNamespace.on('connection', (socket) => {
console.log(`Socket接続: ${socket.id}`);
// ユーザー参加処理
socket.on(
'join-room',
(data: { username: string; room: string }) => {
this.handleUserJoin(socket, data);
}
);
// メッセージ送信処理
socket.on(
'send-message',
(data: { content: string }) => {
this.handleMessageSend(socket, data);
}
);
// プライベートメッセージ
socket.on(
'private-message',
(data: {
targetUserId: string;
content: string;
}) => {
this.handlePrivateMessage(socket, data);
}
);
// タイピング状態
socket.on('typing-start', () => {
this.handleTypingStart(socket);
});
socket.on('typing-stop', () => {
this.handleTypingStop(socket);
});
// ルーム変更
socket.on(
'change-room',
(data: { newRoom: string }) => {
this.handleRoomChange(socket, data);
}
);
// 切断処理
socket.on('disconnect', () => {
this.handleUserDisconnect(socket);
});
// エラーハンドリング
socket.on('error', (error) => {
console.error(
`Socket エラー [${socket.id}]:`,
error
);
});
});
}
private handleUserJoin(
socket: any,
data: { username: string; room: string }
): void {
const { username, room } = data;
// ユーザー情報を保存
const user: User = {
id: socket.id,
username,
room,
joinedAt: new Date(),
};
this.users.set(socket.id, user);
// ルームに参加
socket.join(room);
this.addUserToRoom(room, socket.id);
// 参加通知をルームにブロードキャスト
socket.to(room).emit('user-joined', {
user: { id: user.id, username: user.username },
message: `${username}さんが参加しました`,
timestamp: new Date(),
});
// クライアントに参加確認を送信
socket.emit('join-confirmed', {
room,
users: this.getRoomUsers(room),
});
console.log(
`ユーザー参加: ${username} → ルーム: ${room}`
);
}
private handleMessageSend(
socket: any,
data: { content: string }
): void {
const user = this.users.get(socket.id);
if (!user) return;
const message: ChatMessage = {
id: `${socket.id}-${Date.now()}`,
username: user.username,
content: data.content,
room: user.room,
timestamp: new Date(),
};
// ルーム全体にメッセージをブロードキャスト
this.io
.of('/chat')
.to(user.room)
.emit('new-message', message);
console.log(
`メッセージ [${user.room}] ${user.username}: ${data.content}`
);
}
private handlePrivateMessage(
socket: any,
data: { targetUserId: string; content: string }
): void {
const sender = this.users.get(socket.id);
const recipient = this.users.get(data.targetUserId);
if (!sender || !recipient) {
socket.emit('error', {
message: 'ユーザーが見つかりません',
});
return;
}
const privateMessage = {
id: `private-${Date.now()}`,
from: { id: sender.id, username: sender.username },
content: data.content,
timestamp: new Date(),
};
// 送信者と受信者にメッセージを送信
socket.emit('private-message-sent', privateMessage);
socket
.to(data.targetUserId)
.emit('private-message-received', privateMessage);
console.log(
`プライベートメッセージ: ${sender.username} → ${recipient.username}`
);
}
private handleTypingStart(socket: any): void {
const user = this.users.get(socket.id);
if (!user) return;
socket.to(user.room).emit('user-typing', {
userId: user.id,
username: user.username,
});
}
private handleTypingStop(socket: any): void {
const user = this.users.get(socket.id);
if (!user) return;
socket.to(user.room).emit('user-stopped-typing', {
userId: user.id,
username: user.username,
});
}
private handleRoomChange(
socket: any,
data: { newRoom: string }
): void {
const user = this.users.get(socket.id);
if (!user) return;
const oldRoom = user.room;
// 古いルームから退出
socket.leave(oldRoom);
this.removeUserFromRoom(oldRoom, socket.id);
// 退出通知
socket.to(oldRoom).emit('user-left', {
user: { id: user.id, username: user.username },
message: `${user.username}さんが退出しました`,
});
// 新しいルームに参加
user.room = data.newRoom;
socket.join(data.newRoom);
this.addUserToRoom(data.newRoom, socket.id);
// 参加通知
socket.to(data.newRoom).emit('user-joined', {
user: { id: user.id, username: user.username },
message: `${user.username}さんが参加しました`,
});
// クライアントに変更確認
socket.emit('room-changed', {
oldRoom,
newRoom: data.newRoom,
users: this.getRoomUsers(data.newRoom),
});
console.log(
`ルーム変更: ${user.username} ${oldRoom} → ${data.newRoom}`
);
}
private handleUserDisconnect(socket: any): void {
console.log(
`セキュア切断: ${socket.username} [${socket.id}]`
);
// セッション情報のクリーンアップ
// ログ記録
}
// ルーム管理のヘルパーメソッド
private setupRoomManagement(): void {
// 定期的なルーム状況のブロードキャスト
setInterval(() => {
this.io.of('/chat').emit('room-stats', {
totalUsers: this.users.size,
totalRooms: this.rooms.size,
timestamp: new Date(),
});
}, 30000); // 30秒ごと
}
private addUserToRoom(
room: string,
userId: string
): void {
if (!this.rooms.has(room)) {
this.rooms.set(room, new Set());
}
this.rooms.get(room)!.add(userId);
}
private removeUserFromRoom(
room: string,
userId: string
): void {
const roomUsers = this.rooms.get(room);
if (roomUsers) {
roomUsers.delete(userId);
if (roomUsers.size === 0) {
this.rooms.delete(room);
}
}
}
private getRoomUsers(room: string): any[] {
const roomUsers = this.rooms.get(room);
if (!roomUsers) return [];
return Array.from(roomUsers)
.map((userId) => {
const user = this.users.get(userId);
return user
? { id: user.id, username: user.username }
: null;
})
.filter(Boolean);
}
public start(port: number = 3000): void {
this.server.listen(port, () => {
console.log(
`Socket.IOサーバーがポート${port}で起動しました`
);
console.log(`http://localhost:${port}`);
});
}
}
// サーバー起動
const chatServer = new SocketIOChatServer();
chatServer.start(3000);
認証とセキュリティの実装
WebSocket 通信におけるセキュリティ対策は非常に重要です。
JWT 認証の実装
typescript// middleware/auth.ts
import jwt from 'jsonwebtoken';
import { Socket } from 'socket.io';
interface AuthenticatedSocket extends Socket {
userId?: string;
username?: string;
}
interface JWTPayload {
userId: string;
username: string;
iat: number;
exp: number;
}
const JWT_SECRET =
process.env.JWT_SECRET || 'your-secret-key';
// JWT認証ミドルウェア
export const authenticateSocket = (
socket: AuthenticatedSocket,
next: (err?: Error) => void
) => {
try {
// トークンの取得(クエリパラメータまたはハンドシェイクから)
const token =
socket.handshake.auth.token ||
(socket.handshake.query.token as string);
if (!token) {
throw new Error('認証トークンがありません');
}
// トークンの検証
const decoded = jwt.verify(
token,
JWT_SECRET
) as JWTPayload;
// ソケットにユーザー情報を追加
socket.userId = decoded.userId;
socket.username = decoded.username;
console.log(
`認証成功: ${decoded.username} [${socket.id}]`
);
next();
} catch (error) {
console.error('Socket認証エラー:', error);
next(new Error('認証に失敗しました'));
}
};
// レート制限ミドルウェア
class RateLimiter {
private requests: Map<string, number[]> = new Map();
private readonly maxRequests: number;
private readonly windowMs: number;
constructor(
maxRequests: number = 100,
windowMs: number = 60000
) {
this.maxRequests = maxRequests;
this.windowMs = windowMs;
}
public checkLimit(socketId: string): boolean {
const now = Date.now();
const requests = this.requests.get(socketId) || [];
// 古いリクエストを除去
const validRequests = requests.filter(
(timestamp) => now - timestamp < this.windowMs
);
if (validRequests.length >= this.maxRequests) {
return false; // レート制限に達している
}
// 新しいリクエストを記録
validRequests.push(now);
this.requests.set(socketId, validRequests);
return true;
}
public cleanup(): void {
const now = Date.now();
this.requests.forEach((timestamps, socketId) => {
const validRequests = timestamps.filter(
(timestamp) => now - timestamp < this.windowMs
);
if (validRequests.length === 0) {
this.requests.delete(socketId);
} else {
this.requests.set(socketId, validRequests);
}
});
}
}
export const rateLimiter = new RateLimiter();
// 定期クリーンアップ
setInterval(() => {
rateLimiter.cleanup();
}, 60000); // 1分ごと
セキュアな WebSocket サーバー
typescript// src/secure-websocket-server.ts
import { Server } from 'socket.io';
import {
authenticateSocket,
rateLimiter,
} from '../middleware/auth';
import rateLimit from 'express-rate-limit';
import helmet from 'helmet';
class SecureWebSocketServer {
private io: Server;
constructor(server: any) {
this.io = new Server(server, {
cors: {
origin:
process.env.ALLOWED_ORIGINS?.split(',') ||
'http://localhost:3000',
credentials: true,
},
// セキュリティ設定
pingTimeout: 60000,
pingInterval: 25000,
upgradeTimeout: 10000,
maxHttpBufferSize: 1e6, // 1MB
allowEIO3: false, // 古いバージョンを無効化
});
this.setupSecurityMiddleware();
this.setupEventHandlers();
}
private setupSecurityMiddleware(): void {
// 認証ミドルウェア
this.io.use(authenticateSocket);
// カスタムレート制限
this.io.use((socket, next) => {
if (!rateLimiter.checkLimit(socket.id)) {
next(new Error('レート制限に達しました'));
return;
}
next();
});
// IPアドレスの検証
this.io.use((socket, next) => {
const clientIP = socket.handshake.address;
// ブラックリストのチェック(例)
const blacklistedIPs =
process.env.BLACKLISTED_IPS?.split(',') || [];
if (blacklistedIPs.includes(clientIP)) {
console.warn(
`ブラックリストIPからの接続をブロック: ${clientIP}`
);
next(new Error('アクセスが拒否されました'));
return;
}
console.log(`接続許可: ${clientIP} [${socket.id}]`);
next();
});
}
private setupEventHandlers(): void {
this.io.on('connection', (socket: any) => {
console.log(
`セキュア接続: ${socket.username} [${socket.id}]`
);
// メッセージの検証とサニタイゼーション
socket.on('send-message', (data: any) => {
if (!this.validateMessage(data)) {
socket.emit('error', {
message: '無効なメッセージ形式です',
});
return;
}
// XSS対策:HTMLタグの無効化
const sanitizedMessage = this.sanitizeMessage(
data.content
);
// メッセージ送信処理
this.handleSecureMessage(socket, {
...data,
content: sanitizedMessage,
});
});
// ファイルアップロードの処理
socket.on('upload-file', (data: any) => {
if (!this.validateFileUpload(data)) {
socket.emit('error', {
message: '無効なファイルです',
});
return;
}
this.handleFileUpload(socket, data);
});
// セッション管理
socket.on('disconnect', () => {
this.handleSecureDisconnect(socket);
});
});
}
private validateMessage(data: any): boolean {
// メッセージの基本検証
if (!data || typeof data.content !== 'string') {
return false;
}
// 長さ制限
if (data.content.length > 1000) {
return false;
}
// 不正な文字の検出
const dangerousPatterns = [
/<script/i,
/javascript:/i,
/on\w+=/i,
/<iframe/i,
];
return !dangerousPatterns.some((pattern) =>
pattern.test(data.content)
);
}
private sanitizeMessage(content: string): string {
return content
.replace(/</g, '<')
.replace(/>/g, '>')
.replace(/"/g, '"')
.replace(/'/g, ''')
.replace(/\//g, '/');
}
private validateFileUpload(data: any): boolean {
if (!data.fileName || !data.fileData) {
return false;
}
// ファイル拡張子のチェック
const allowedExtensions = [
'.jpg',
'.jpeg',
'.png',
'.gif',
'.pdf',
'.txt',
];
const fileExtension = data.fileName
.toLowerCase()
.substring(data.fileName.lastIndexOf('.'));
if (!allowedExtensions.includes(fileExtension)) {
return false;
}
// ファイルサイズ制限(5MB)
if (data.fileData.length > 5 * 1024 * 1024) {
return false;
}
return true;
}
private handleSecureMessage(
socket: any,
data: any
): void {
// セキュアなメッセージ処理
// レート制限チェック
if (!rateLimiter.checkLimit(socket.id)) {
socket.emit('error', {
message: 'メッセージ送信頻度が高すぎます',
});
return;
}
// メッセージをブロードキャスト
this.io.emit('new-message', {
id: `${socket.id}-${Date.now()}`,
username: socket.username,
content: data.content,
timestamp: new Date(),
verified: true,
});
}
private handleFileUpload(socket: any, data: any): void {
// ファイルアップロード処理
console.log(
`ファイルアップロード: ${data.fileName} from ${socket.username}`
);
// ここで実際のファイル保存処理を実装
// AWS S3、Google Cloud Storage等への安全なアップロード
}
private handleSecureDisconnect(socket: any): void {
console.log(
`セキュア切断: ${socket.username} [${socket.id}]`
);
// セッション情報のクリーンアップ
// ログ記録
}
}
export { SecureWebSocketServer };
パフォーマンス最適化とエラーハンドリング
WebSocket サーバーの性能向上とエラー対応のベストプラクティスです。
パフォーマンス最適化
typescript// utils/performance-optimizer.ts
import { Server } from 'socket.io';
import Redis from 'ioredis';
class WebSocketPerformanceOptimizer {
private redis: Redis;
private messageQueue: Map<string, any[]> = new Map();
private batchSize: number = 10;
private batchInterval: number = 100; // 100ms
constructor() {
// Redis接続(本番環境での複数サーバー対応)
this.redis = new Redis({
host: process.env.REDIS_HOST || 'localhost',
port: parseInt(process.env.REDIS_PORT || '6379'),
retryDelayOnFailover: 100,
enableReadyCheck: false,
maxRetriesPerRequest: null,
});
this.setupBatchProcessing();
}
// メッセージのバッチ処理
private setupBatchProcessing(): void {
setInterval(() => {
this.processBatchedMessages();
}, this.batchInterval);
}
public queueMessage(room: string, message: any): void {
if (!this.messageQueue.has(room)) {
this.messageQueue.set(room, []);
}
const queue = this.messageQueue.get(room)!;
queue.push(message);
// バッチサイズに達したら即座に処理
if (queue.length >= this.batchSize) {
this.processBatchedMessages(room);
}
}
private processBatchedMessages(
specificRoom?: string
): void {
const roomsToProcess = specificRoom
? [specificRoom]
: Array.from(this.messageQueue.keys());
roomsToProcess.forEach((room) => {
const messages = this.messageQueue.get(room);
if (!messages || messages.length === 0) return;
// バッチでメッセージを送信
this.sendBatchedMessages(room, messages);
// キューをクリア
this.messageQueue.set(room, []);
});
}
private sendBatchedMessages(
room: string,
messages: any[]
): void {
// Socket.IOサーバーインスタンスへの参照が必要
// 実際の実装では、このクラスにServerインスタンスを注入する
console.log(
`バッチ送信: ルーム${room} - ${messages.length}件のメッセージ`
);
}
// 接続プールの最適化
public optimizeConnectionPool(io: Server): void {
// 接続プールサイズの動的調整
const connectionCount = io.engine.clientsCount;
if (connectionCount > 1000) {
// 高負荷時の最適化
io.engine.generateId = () => {
return `optimized-${Math.random()
.toString(36)
.substr(2, 9)}`;
};
}
}
// メモリ使用量の監視
public monitorMemoryUsage(): void {
const memUsage = process.memoryUsage();
if (memUsage.heapUsed > 500 * 1024 * 1024) {
// 500MB
console.warn('高メモリ使用量を検出:', {
heapUsed: `${Math.round(
memUsage.heapUsed / 1024 / 1024
)}MB`,
heapTotal: `${Math.round(
memUsage.heapTotal / 1024 / 1024
)}MB`,
external: `${Math.round(
memUsage.external / 1024 / 1024
)}MB`,
});
// ガベージコレクションの強制実行
if (global.gc) {
global.gc();
}
}
}
// Redis Adapter使用時のクラスター対応
public setupRedisAdapter(io: Server): void {
const {
createAdapter,
} = require('@socket.io/redis-adapter');
const pubClient = this.redis;
const subClient = pubClient.duplicate();
io.adapter(createAdapter(pubClient, subClient));
console.log(
'Redis Adapterが設定されました(クラスター対応)'
);
}
}
export { WebSocketPerformanceOptimizer };
本番環境でのデプロイと運用
WebSocket アプリケーションを本番環境で安全に運用するための設定です。
Docker コンテナ化
dockerfile# Dockerfile
FROM node:18-alpine
WORKDIR /app
# 依存関係のコピーとインストール
COPY package.json yarn.lock ./
RUN yarn install --frozen-lockfile --production
# アプリケーションコードのコピー
COPY . .
# TypeScriptのビルド
RUN yarn build
# 非rootユーザーの作成
RUN addgroup -g 1001 -S nodejs
RUN adduser -S websocket -u 1001
# ユーザー権限の変更
USER websocket
# ポートの公開
EXPOSE 3000
# ヘルスチェック
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
CMD node healthcheck.js
# アプリケーションの開始
CMD ["node", "dist/server.js"]
yaml# docker-compose.yml
version: '3.8'
services:
websocket-app:
build: .
ports:
- '3000:3000'
environment:
- NODE_ENV=production
- JWT_SECRET=${JWT_SECRET}
- REDIS_HOST=redis
- REDIS_PORT=6379
depends_on:
- redis
restart: unless-stopped
networks:
- websocket-network
redis:
image: redis:7-alpine
ports:
- '6379:6379'
command: redis-server --appendonly yes
volumes:
- redis-data:/data
restart: unless-stopped
networks:
- websocket-network
nginx:
image: nginx:alpine
ports:
- '80:80'
- '443:443'
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf
- ./ssl:/etc/nginx/ssl
depends_on:
- websocket-app
restart: unless-stopped
networks:
- websocket-network
volumes:
redis-data:
networks:
websocket-network:
driver: bridge
監視とログ管理
typescript// utils/monitoring.ts
import winston from 'winston';
import { Server } from 'socket.io';
class WebSocketMonitoring {
private logger: winston.Logger;
private metrics: Map<string, number> = new Map();
constructor() {
this.setupLogger();
this.setupMetricsCollection();
}
private setupLogger(): void {
this.logger = winston.createLogger({
level: 'info',
format: winston.format.combine(
winston.format.timestamp(),
winston.format.errors({ stack: true }),
winston.format.json()
),
defaultMeta: { service: 'websocket-server' },
transports: [
new winston.transports.File({
filename: 'logs/error.log',
level: 'error',
}),
new winston.transports.File({
filename: 'logs/combined.log',
}),
new winston.transports.Console({
format: winston.format.simple(),
}),
],
});
}
private setupMetricsCollection(): void {
// 基本メトリクスの初期化
this.metrics.set('connections', 0);
this.metrics.set('messages_sent', 0);
this.metrics.set('messages_received', 0);
this.metrics.set('errors', 0);
// 定期的なメトリクス出力
setInterval(() => {
this.outputMetrics();
}, 60000); // 1分ごと
}
public trackConnection(io: Server): void {
io.on('connection', (socket) => {
this.incrementMetric('connections');
this.logger.info('新規接続', {
socketId: socket.id,
totalConnections: this.metrics.get('connections'),
});
socket.on('disconnect', () => {
this.decrementMetric('connections');
this.logger.info('接続終了', {
socketId: socket.id,
totalConnections: this.metrics.get('connections'),
});
});
socket.onAny(() => {
this.incrementMetric('messages_received');
});
socket.onAnyOutgoing(() => {
this.incrementMetric('messages_sent');
});
socket.on('error', (error) => {
this.incrementMetric('errors');
this.logger.error('Socket エラー', {
socketId: socket.id,
error: error.message,
});
});
});
}
private incrementMetric(key: string): void {
const current = this.metrics.get(key) || 0;
this.metrics.set(key, current + 1);
}
private decrementMetric(key: string): void {
const current = this.metrics.get(key) || 0;
this.metrics.set(key, Math.max(0, current - 1));
}
private outputMetrics(): void {
const metricsData = Object.fromEntries(this.metrics);
this.logger.info('WebSocket メトリクス', {
timestamp: new Date(),
metrics: metricsData,
memory: process.memoryUsage(),
uptime: process.uptime(),
});
}
public getMetrics(): Record<string, number> {
return Object.fromEntries(this.metrics);
}
}
export { WebSocketMonitoring };
まとめ
本記事では、Node.js を使った WebSocket 通信の実装について、基本的な概念から実践的なアプリケーション開発まで詳しく解説いたしました。
学習した重要なポイント
# | 項目 | 重要度 | 実装のポイント |
---|---|---|---|
1 | WebSocket と HTTP の違い | ⭐⭐⭐ | リアルタイム通信の必要性理解 |
2 | ws ライブラリの基本実装 | ⭐⭐⭐ | シンプルで軽量な WebSocket 通信 |
3 | Socket.IO を使った高機能実装 | ⭐⭐⭐ | 名前空間、ルーム、フォールバック対応 |
4 | React/Next.js 統合 | ⭐⭐ | フロントエンドでのリアルタイム実装 |
5 | 認証とセキュリティ | ⭐⭐⭐ | JWT 認証、レート制限、XSS 対策 |
6 | パフォーマンス最適化 | ⭐⭐ | バッチ処理、メモリ管理、Redis 連携 |
7 | 本番環境運用 | ⭐⭐ | Docker 化、監視、ログ管理 |
実際のプロダクトでの活用
WebSocket 技術は以下のような場面で実際に活用されています:
- チャットアプリケーション: Slack、Discord、Teams 等のリアルタイムメッセージング
- ライブ配信: YouTube Live、Twitch 等のリアルタイムコメント機能
- 協調編集: Google Docs、Figma 等の同時編集機能
- オンラインゲーム: リアルタイム対戦、マルチプレイヤーゲーム
- ダッシュボード: 株価、システム監視等のリアルタイムデータ表示
- IoT アプリケーション: センサーデータのリアルタイム収集・表示
今後の発展
WebSocket 技術は今後も進化を続けており、以下のような発展が期待されます:
- WebRTC 統合: より低遅延な P2P 通信の実現
- WebAssembly 対応: 高性能なリアルタイム処理の実装
- エッジコンピューティング: CDN レベルでの WebSocket 処理
- 5G 対応: モバイル環境でのより高速なリアルタイム通信
WebSocket の基礎をしっかりと理解し、実践的な実装経験を積むことで、次世代のリアルタイム Web アプリケーション開発において大きなアドバンテージを得ることができるでしょう。ぜひ実際のプロジェクトで活用して、ユーザーに価値のあるリアルタイム体験を提供してください。
関連リンク
- review
今の自分に満足していますか?『持たざる者の逆襲 まだ何者でもない君へ』溝口勇児
- review
ついに語られた業界の裏側!『フジテレビの正体』堀江貴文が描くテレビ局の本当の姿
- review
愛する勇気を持てば人生が変わる!『幸せになる勇気』岸見一郎・古賀史健のアドラー実践編で真の幸福を手に入れる
- review
週末を変えれば年収も変わる!『世界の一流は「休日」に何をしているのか』越川慎司の一流週末メソッド
- review
新しい自分に会いに行こう!『自分の変え方』村岡大樹の認知科学コーチングで人生リセット
- review
科学革命から AI 時代へ!『サピエンス全史 下巻』ユヴァル・ノア・ハラリが予見する人類の未来