導入
Pub/Subパターンは、パブリッシャーがデータ/メッセージを生成し、サブスクライバーが特定の種類のメッセージを受信するためにサブスクライブする、汎用性の高い一方向メッセージングパターンです。ピアツーピアアーキテクチャ、または通信を仲介するメッセージブローカーを使用して実装できます。.
上の画像は、ピアツーピアのPub/Subモデルを示しています。このモデルでは、パブリッシャーが仲介者を介さずにサブスクライバーに直接メッセージを送信します。サブスクライバーは、メッセージを受信するためにパブリッシャーのアドレスまたはエンドポイントを知っている必要があります。.
上の図では、Pub/Sub モデルはメッセージブローカーを中央ハブとして利用し、パブリッシャーとサブスクライバー間のメッセージ送信を行います。ブローカーはメッセージ交換を仲介し、パブリッシャーからサブスクライバーへのメッセージを配信します。サブスクライバーノードは、パブリッシャーではなくブローカーに直接サブスクライブします。ブローカーの存在により、パブリッシャーとサブスクライバーの両方がブローカーとのみやり取りするため、システムノードの分離性が向上します。このチュートリアルでは、リアルタイムチャットアプリケーションを構築し、このパターンをさらに実証します。.
前提条件
- Node.js (バージョン >= 12) がオペレーティング システムにインストールされています。.
- VSCodeのようなコードエディタ
- Redis がマシンにインストールされています。
- HTML、DOM、VanillaJS、WebSocket の基本的な理解。.
ステップ1 – サーバー側の実装
サーバー側の実行を開始するには、初期コマンドを使用して基本的な Nodejs アプリケーションを初期化します。
npm init -y上記のコマンドは、デフォルトの package.json ファイルを作成します。.
次に、このビルド全体を通じて必要となる WebSocket (ws) 依存パッケージをインストールします。
npm install wsサーバーサイド実装は、シンプルなサーバーサイドチャットアプリケーションです。以下のワークフローに従います。
- サーバーをセットアップする
- HTMLファイルを読み取ってブラウザでレンダリングする
- WebSocket 接続を設定します。.
サーバーのセットアップ
ディレクトリに app.js というファイルを作成し、その中に次のコードを配置します。
const http = require("http");
const server = http.createServer((req, res) => {
res.end("Hello Chat App");
});
const PORT = 3459;
server.listen(PORT, () => {
console.log(`Server up and running on port ${PORT}`);
});方法 サーバーの作成 モジュール内 http 国内 ノード.js これはサーバーの起動に使用されます。サーバーがリクエストをリッスンするポートが設定され、作成されたサーバーインスタンスの listen メソッドが呼び出され、指定されたポートで受信リクエストをリッスンします。.
注文 ノードアプリ.js ターミナルで実行すると、次のような応答が返されます。
OutputServer is up and running on port 3459ブラウザでこのポートを要求すると、次のような応答が返されます。
HTMLファイルを読み取ってブラウザでレンダリングする
メイン フォルダーに index.html というファイルを作成し、次のコードをコピーします。
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>Document</title>
</head>
<body>
<p>Serving HTML file</p>
</body>
</html>これはHelloをレンダリングする基本的なHTMLファイルです。サーバーにHTTPリクエストが送信されるたびに、このファイルを読み込み、レスポンスとしてレンダリングする必要があります。.
// app.js
const server = http.createServer((req, res) => {
const htmlFilePath = path.join(__dirname, "index.html");
fs.readFile(htmlFilePath, (err, data) => {
if (err) {
res.writeHead(500);
res.end("Error occured while reading file");
}
res.writeHead(200, { "Content-Type": "text/html" });
res.end(data);
});
});ここでは、内部パスモジュールと関数を使用します。 参加する パスセグメントを結合するために使用します。次に関数を使用します ファイル読み取り ファイルを読むには インデックス.html 非同期的に使用されます。2つの引数が必要です：読み込むファイルへのパスとリードバック。ステータスコード 500 レスポンスヘッダーが送信され、エラーメッセージがクライアントに送信されます。データが正常に読み取られた場合は、成功ステータスコードが返されます。 200 レスポンスヘッダーを送信し、レスポンスデータ（この場合はファイルの内容）をクライアントに送信します。UTF-8などのエンコードが指定されていない場合は、生のバッファが返されます。それ以外の場合は、ファイル HTML 返されます。.
ブラウザでサーバーにリクエストすると、次のようになります。
WebSocket接続の設定
// app.js
const webSocketServer = new WebSocket.Server({ server });
webSocketServer.on("connection", (client) => {
console.log("successfully connected to the client");
client.on("message", (streamMessage) => {
console.log("message", streamMessage);
distributeClientMessages(streamMessage);
});
});
const distributeClientMessages = (message) => {
for (const client of webSocketServer.clients) {
if (client.readyState === WebSocket.OPEN) {
client.send(message);
}
}
};前のコード行では、新しいWebSocketサーバーを作成しました。 ウェブソケットサーバー 作成してサーバーに送信します。 HTTP 既存のポートに接続します。これにより、標準のHTTPリクエストとWebSocket接続の両方を単一のポート3459で処理できるようになります。.
on() 接続イベントは、WebSocket 接続が確立されたときに発生します。関数内のクライアントは 折り返し電話 WebSocket接続オブジェクトは、クライアントへの接続を表します。メッセージの送受信や、クライアントメッセージなどのイベントのリッスンに使用されます。.
ここで、distributeClientMessages関数は、接続中のすべてのクライアントに受信メッセージを送信するために使用されています。この関数はメッセージを引数として受け取り、サーバーに接続しているクライアントを反復処理します。次に、各クライアントの接続状態（readyState === WebSocket.OPEN）を確認します。これは、サーバーが接続が開いているクライアントにのみメッセージを送信できるようにするためです。クライアント接続が開いている場合、サーバーはclient.send(message)メソッドを使用してそのクライアントにメッセージを送信します。.
ステップ2 – クライアント側の実装
クライアント側を実行するには、ファイル インデックス.html 私たち自身も少し変わります。.
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>Document</title>
</head>
<body>
<p>Pub/Sub Pattern with Chat Messaging</p>
<div id="messageContainer"></div>
<form id="messageForm">
<form id="messageForm">
<input
type="text"
id="messageText"
placeholder="Send a message"
style="
padding: 10px;
margin: 5px;
border-radius: 5px;
border: 1px solid #ccc;
outline: none;
"
onfocus="this.style.borderColor='#007bff';"
onblur="this.style.borderColor='#ccc';"
/>
<input
type="button"
value="Send Message"
style="
padding: 10px;
margin: 5px;
border-radius: 5px;
background-color: #007bff;
color: white;
border: none;
cursor: pointer;
"
onmouseover="this.style.backgroundColor='#0056b3';"
onmouseout="this.style.backgroundColor='#007bff';"
/>
</form>
</form>
<script>
const url = window.location.host;
const socket = new WebSocket(`ws://${url}`);
</script>
</body>
</html>このコードスニペットでは、入力欄とメッセージを送信するためのボタンを持つフォーム要素を追加しました。WebSocket接続はクライアントによって開始されます。最初にセットアップしたWebSocketサーバーと通信するには、ws://url（使用するサーバー）を指定したWebSocketオブジェクトのインスタンスを作成する必要があります。ログインすると、URL変数にはサーバーがリッスンしているポート3459への接続URLが格納されます。.
// app.js
console.log("url", url); // localhost:3459
console.log("socket", socket); // { url: "ws://localhost:3459/", readyState: 0, bufferedAmount: 0, onopen: null, onerror: null, onclose: null, extensions: "", protocol: "", onmessage: null, binaryType: "blob" }したがって、ブラウザでサーバーにリクエストを送信すると、次のように表示されます。
クライアントからサーバーにメッセージを送信し、サーバーからメッセージを受信できるようにスクリプトをアップグレードしましょう。.
// index.html
<script>
const url = window.location.host;
const socket = new WebSocket(`ws://${url}`);
const messageContainer = document.getElementById("messageContainer");
socket.onmessage = function (eventMessage) {
eventMessage.data.text().then((text) => {
const messageContent = document.createElement("p");
messageContent.innerHTML = text;
document
.getElementById("messageContainer")
.appendChild(messageContent);
});
};
const form = document.getElementById("messageForm");
form.addEventListener("submit", (event) => {
event.preventDefault();
const message = document.getElementById("messageText").value;
socket.send(message);
document.getElementById("messageText").value = "";
});
</script>前述のように、クライアント（ブラウザ）からサーバーにリクエストを送信するURLを取得し、そのURLを使用して新しいWebSocketオブジェクトのインスタンスを作成します。そして、ボタンをクリックすると、 メッセージを送信 フォーム要素にイベントを作成します。ユーザーがUIに入力したテキストは入力要素から抽出され、ソケットインスタンスのsendメソッドが呼び出されてサーバーに送信されます。.
イベント オンメッセージ ソケットオブジェクトで呼び出され、サーバーからメッセージを受信したときに実行されます。受信メッセージのユーザーインターフェースを更新するために使用されます。パラメータ イベントメッセージ 機能上 折り返し電話 サーバーから送信されたデータ（メッセージ）はBlobとして返されます。その後、Blobデータに対してtext()メソッドが使用され、Promiseが返されます。then()メソッドを使用して解決することで、サーバーから実際のテキストが取得されます。.
作成したものをテストしてみましょう。以下を実行してサーバーを起動します。
node app.js次に、2つの異なるブラウザタブを開き、 http://localhost:3459/ 開いて、タブ間でメッセージを送信してテストします。
ステップ3 – アプリケーションのスケール
アプリケーションが成長し始め、チャットサーバーのインスタンスを複数用意して拡張しようとしたとします。目指すのは、2つの異なるサーバーに接続している2人のユーザーが、互いにテキストメッセージを送信できるようにすることです。現在はサーバーが1つしかなく、例えば別のサーバーへの接続をリクエストすると、 http://localhost:3460/、ポート上のサーバーメッセージ 3459 ありません。つまり、接続しているユーザーのみが 3460 自分自身とチャットすることができます。現在の実装では、実行中のサーバーインスタンスにチャットメッセージが送信されると、そのメッセージは特定のサーバーに接続しているクライアントにのみローカルに配信されます。 http://localhost:3459/ 2つの異なるブラウザで開きます。では、2つの異なるサーバーを統合して相互に通信できるようにする方法を見てみましょう。.
ステップ4 – メッセージブローカーとしてのRedis
Redisは、高速で柔軟なインメモリデータ構造ストアです。データベースやキャッシュサーバーとしてデータを保存するのによく使用されます。さらに、集中型のPub/Subメッセージ交換パターンを実装するためにも使用できます。Redisの高速性と柔軟性により、分散システムにおけるデータ共有の選択肢として非常に人気があります。.
ここでの目標は、Redisをメッセージブローカーとして利用し、チャットサーバーを統合することです。各サーバーインスタンスは、クライアント（ブラウザ）から受信したすべてのメッセージを同時にメッセージブローカーにパブリッシュします。メッセージブローカーは、サーバーインスタンスから送信されるすべてのメッセージをサブスクライブします。.
ファイルしましょう アプリ.js 自分自身を変えましょう:
//app.js
const http = require("http");
const fs = require("fs");
const path = require("path");
const WebSocket = require("ws");
const Redis = require("ioredis");
const redisPublisher = new Redis();
const redisSubscriber = new Redis();
const server = http.createServer((req, res) => {
const htmlFilePath = path.join(__dirname, "index.html");
fs.readFile(htmlFilePath, (err, data) => {
if (err) {
res.writeHead(500);
res.end("Error occured while reading file");
}
res.writeHead(200, { "Content-Type": "text/html" });
res.end(data);
});
});
const webSocketServer = new WebSocket.Server({ server });
webSocketServer.on("connection", (client) => {
console.log("succesfully connected to the client");
client.on("message", (streamMessage) => {
redisPublisher.publish("chat_messages", streamMessage);
});
});
redisSubscriber.subscribe("chat_messages");
console.log("sub", redisSubscriber.subscribe("messages"));
redisSubscriber.on("message", (channel, message) => {
console.log("redis", channel, message);
for (const client of webSocketServer.clients) {
if (client.readyState === WebSocket.OPEN) {
client.send(message);
}
}
});
const PORT = process.argv[2] || 3459;
server.listen(PORT, () => {
console.log(`Server up and running on port ${PORT}`);
});ここでは、公開/共有機能を使用します。 レディス 2つの異なる接続インスタンスを使用します。1つはメッセージのパブリッシュ用、もう1つはチャネルのサブスクライブ用です。クライアントからメッセージが送信されると、redisPublisherインスタンスのpublisherメソッドを使用して、redisPublisherというRedisチャネルにパブリッシュされます。 "「チャットメッセージ」" パブリッシュします。redisSubscribeインスタンスのsubscribeメソッドを呼び出して、chat_messageと同じチャネルをサブスクライブします。このチャネルにメッセージがパブリッシュされるたびに、redisSubscriber.onイベントリスナーがトリガーされます。このイベントリスナーは、現在接続中のすべてのWebSocketクライアントで繰り返し実行され、受信したメッセージを各クライアントに送信します。これにより、ユーザーがメッセージを送信すると、各サーバーインスタンスに接続している他のすべてのユーザーがそのメッセージをリアルタイムで受信できるようになります。.
2 つの異なるサーバーをセットアップする場合は、次のようにします。
node app.js 3459
node app.js 3460インスタンス上でチャットテキストが送信された際に、特定のサーバーだけではなく、接続されたサーバー全体にメッセージをブロードキャストできるようになりました。これを行うには、次のコマンドを実行します。 http://localhost:3459/ そして http://localhost:3460/ テストして、それらの間でチャットを送信し、2 つのサーバー間でメッセージがリアルタイムでブロードキャストされるのを確認します。.
チャンネルで公開されたメッセージは以下から閲覧できます。 redis-cli 定期的にメッセージを受け取るには、チャンネルを監視し、登録してください。
注文 redis-cli 実行します。次に入力します モニター ブラウザに戻ってチャットを始めましょう。ターミナルには、Wowチャットテキストを送信した場合、次のような画面が表示されるはずです。
共有公開メッセージを表示するには、同じコマンドを使用します redis-cli 走って チャンネル名を購読する 私たちについてのチャンネル名を入力してください。 チャットメッセージ テキストを送信すると、ターミナルに次のようなものが表示されます。ブラウザからは次のように表示されます。
これで、異なるポートまたは異なるマシンでサーバーの複数のインスタンスを実行できるようになりました。同じ Redis チャネルにサブスクライブしている限り、接続されているすべてのクライアントにメッセージを受信してブロードキャストできるため、ユーザーはインスタンス間でシームレスにチャットできます。.
はじめに、メッセージブローカーを使用して Pub/Sub パターンを実装する方法について説明しましたね。この例はそれを完璧に要約しています。.
上の図では、2つの異なるクライアント（ブラウザ）がチャットサーバーに接続されています。チャットサーバーは直接接続されておらず、Redisインスタンスを介して接続されています。つまり、クライアント接続はそれぞれ独立して処理されますが、情報（チャットメッセージ）は共通の媒体（Redis）を介して共有されます。上記の各チャットサーバーはRedisに接続します。この接続は、Redisへのメッセージのパブリッシュと、Redisチャネルへのサブスクライブに使用されます。ユーザーがメッセージを送信すると、チャットサーバーはRedis内の指定されたチャネルにメッセージをパブリッシュします。.
Redis はパブリッシュされたメッセージを受信すると、参加しているすべてのチャットサーバーにメッセージをブロードキャストします。各チャットサーバーは、接続しているすべてのクライアントにメッセージを送信します。これにより、どのサーバーに接続しているかに関係なく、すべてのユーザーが送信したメッセージを受信できるようになります。.
このアーキテクチャにより、必要に応じてサーバーインスタンスを追加することで、チャットアプリケーションを水平方向にスケールできます。Redisのパブリッシュ/サブスクライブシステムにより、すべてのインスタンス間で均一なメッセージ配信が保証されるため、各インスタンスは接続されたクライアント群を個別に管理できます。この構成は、多数の同時ユーザーを効率的に処理し、アプリケーションの可用性を確保します。.
結果
このチュートリアルでは、パブリッシュ/サブスクライブ・パターンについて学び、Redisをメッセージブローカーとして用いたシンプルなチャットアプリケーションを作成し、このパターンを実証しました。次のステップでは、メッセージブローカーが最適なソリューションではない場合、例えば単一障害点（ブローカー）が許容されない複雑な分散システムなどにおいて、ピアツーピア・メッセージング・システムを実装する方法を学びます。.
