Node.js의 게시/구독 패턴
저자는 Write for DOnations 프로그램의 일환으로 No Kid Hungry를 기부 대상으로 선택했습니다.
소개
Pub/Sub 패턴은 게시자가 데이터/메시지를 생성하고 구독자가 특정 유형의 메시지를 수신하도록 등록하는 다목적 단방향 메시징 패턴입니다. 통신을 중재하기 위해 P2P 아키텍처나 메시지 브로커를 사용하여 구현할 수 있습니다.
위 이미지는 게시자가 중재자 없이 구독자에게 직접 메시지를 보내는 P2P Pub/Sub 모델을 보여줍니다. 구독자는 메시지를 받으려면 게시자의 주소나 끝점을 알아야 합니다.
참고: 이 경우 노드는 일반적으로 메시징 네트워크의 활성 참가자를 의미하며 정보를 게시하는 서비스이거나 정보를 수신하는 서비스(구독자)일 수 있습니다.
위 이미지에서 Pub/Sub 모델은 메시지 브로커를 중앙 허브로 사용하여 게시자와 구독자 간에 메시지를 전달합니다. 브로커는 메시지 교환을 중재하여 게시자의 메시지를 구독자에게 배포합니다. 구독자 노드는 게시자가 아닌 브로커를 직접 구독합니다.
게시자와 구독자 모두 브로커하고만 상호 작용하므로 브로커가 있으면 시스템 노드 간의 분리가 향상됩니다.
이 튜토리얼에서는 이 패턴을 추가로 보여주기 위해 실시간 채팅 애플리케이션을 구축합니다.
전제조건
- 운영 체제에 Node.js(버전 >= 12)가 설치되어 있습니다. Node.js 설치 및 로컬 개발 환경 생성 방법을 따라갈 수 있습니다.
- VSCode와 같은 코드 편집기.
- Redis가 컴퓨터에 설치되어 있습니다. 컴퓨터에 Redis를 설치하는 방법에 대한 공식 문서를 따를 수 있습니다.
- HTML, DOM, VanillaJS 및 WebSocket에 대한 기본 이해.
1단계 - 서버측 구현
서버 측 구현을 시작하기 위해 다음 명령을 사용하여 기본 Nodejs 앱을 초기화합니다.
npm init -y
위 명령은 기본 package.json 파일을 생성합니다.
package.json 파일은 Node.js 프로젝트의 핵심 구성 요소입니다. 이는 프로젝트 이름, 버전, 종속성, 스크립트 등과 같은 다양한 메타데이터를 포함하는 프로젝트의 매니페스트 역할을 합니다. npm install 또는 yarn add를 사용하여 프로젝트에 종속성을 추가하면 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}`);
});
Node.js의 내장 http 모듈에 있는 createServer 메소드는 서버를 설정하는 데 사용됩니다. 서버가 요청을 수신해야 하는 PORT가 설정되었으며, 지정된 포트에서 들어오는 요청을 수신하기 위해 생성된 서버 인스턴스에서 수신 메소드가 호출되었습니다.
터미널에서 node app.js 명령을 실행하면 다음과 같은 응답을 받게 됩니다.
Server 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 요청이 있을 때마다 응답으로 제공해야 합니다.
const http = require("http");
const fs = require("fs");
const path = require("path");
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);
});
});
여기서는 내장된 경로 모듈과 조인 함수를 사용하여 경로 세그먼트를 함께 연결합니다. 그런 다음 readFile 함수를 사용하여 index.html 파일을 비동기적으로 읽습니다. 읽을 파일의 경로와 콜백이라는 두 가지 인수를 사용합니다. 500 상태 코드가 응답 헤더로 전송되고 오류 메시지가 클라이언트로 다시 전송됩니다.
데이터가 성공적으로 읽혀지면 200 성공 상태 코드를 응답 헤더로 보내고 응답 데이터(이 경우 파일의 내용)를 클라이언트에 다시 보냅니다. UTF-8 인코딩과 같이 인코딩이 지정되지 않으면 원시 버퍼가 반환됩니다. 그렇지 않으면 HTML 파일이 반환됩니다.
브라우저에서 서버에 요청하면 다음이 표시됩니다.
WebSocket 연결 설정
const WebSocket = require("ws");
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 서버인 webSocketServer를 생성하고 이를 기존 HTTP 서버에 연결합니다. 이를 통해 동일한 포트 3459에서 표준 HTTP 요청과 WebSocket 연결을 모두 처리할 수 있습니다.
WebSocket 연결이 성공적으로 설정되면 on() 연결 이벤트가 트리거됩니다. 콜백 함수의 클라이언트는 클라이언트에 대한 연결을 나타내는 WebSocket 연결 개체입니다. 메시지를 보내고 받고 클라이언트에서 message와 같은 이벤트를 수신하는 데 사용됩니다.
여기서 distrubuteClientMessages 함수는 수신된 메시지를 연결된 모든 클라이언트에 보내는 데 사용됩니다. message 인수를 받아 서버에 연결된 클라이언트를 반복합니다. 그런 다음 각 클라이언트의 연결 상태를 확인합니다(readyState === WebSocket.OPEN). 이는 서버가 연결이 열려 있는 클라이언트에게만 메시지를 보내도록 하기 위한 것입니다. 클라이언트의 연결이 열려 있으면 서버는 client.send(message) 메서드를 사용하여 해당 클라이언트에 메시지를 보냅니다.
2단계 - 클라이언트 측 구현
클라이언트측 구현을 위해 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>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}`);
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" }
</script>
</body>
</html>
이 코드에서는 입력과 메시지 전송 버튼이 있는 양식 요소를 추가했습니다. WebSocket 연결은 클라이언트에 의해 시작되며 초기에 설정한 WebSocket 지원 서버와 통신하려면 다음을 식별하는 ws://url을 지정하는 WebSocket 객체의 인스턴스를 생성해야 합니다. 우리가 사용하고 싶은 서버. URL 및 소켓 변수는 기록될 때 서버가 포트 3459와 WebSocket 개체에서 각각 수신 대기하는 포트에 대한 URL 연결을 갖게 됩니다.
따라서 브라우저에서 서버에 요청을 입력하면 다음과 같은 내용이 표시됩니다.
클라이언트에서 서버로 메시지를 보내고 서버에서 메시지를 받을 수 있도록 스크립트를 업그레이드해 보겠습니다.
<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 개체 인스턴스를 만듭니다. 그런 다음 메시지 보내기 버튼을 클릭하면 양식 요소에 이벤트가 생성됩니다. 사용자가 사용자 인터페이스에 입력한 텍스트는 입력 요소에서 추출되고 소켓 인스턴스에서 send 메소드가 호출되어 서버에 메시지를 보냅니다.
참고: WebSocket 연결을 통해 서버에 메시지를 보내려면 일반적으로 WebSocket 개체의 send() 메서드가 호출되며 단일 메시지 인수가 필요합니다. , 이는 ArrayBuffer, Blob, string 또는 typed array일 수 있습니다. 이 메서드는 전송될 지정된 메시지를 버퍼링하고 서버에 메시지를 보내기 전에 이를 반환합니다.
소켓 객체에서 호출되는 onmessage 이벤트는 서버로부터 메시지가 수신될 때 트리거됩니다. 이는 수신 메시지의 사용자 인터페이스를 업데이트하는 데 사용됩니다. 콜백 함수의 eventMessage 매개변수에는 서버에서 보낸 데이터(메시지)가 있지만 Blob으로 반환됩니다. 그런 다음 text() 메서드가 Blob 데이터에 사용됩니다. 이 메서드는 약속을 반환하고 then()을 사용하여 확인되어 서버에서 실제 텍스트를 가져옵니다.
우리가 가지고 있는 것을 테스트해 봅시다. 다음을 실행하여 서버를 시작합니다.
node app.js
그런 다음 두 개의 서로 다른 브라우저 탭을 열고 http://localhost:3459/를 연 다음 탭 간에 메시지를 보내 테스트해 보세요.
3단계 - 애플리케이션 확장
애플리케이션이 성장하기 시작하고 채팅 서버의 여러 인스턴스를 보유하여 애플리케이션을 확장하려고 한다고 가정해 보겠습니다. 우리가 달성하고자 하는 것은 서로 다른 두 서버에 연결된 두 명의 서로 다른 사용자가 서로에게 성공적으로 문자 메시지를 보낼 수 있어야 한다는 것입니다. 현재 서버는 하나만 있고 다른 서버(예: http://localhost:3460/)를 요청하면 3459 포트에 있는 서버에 대한 메시지가 없습니다. ; 즉, 3460에 연결된 사용자만 자신과 채팅할 수 있습니다. 현재 구현은 작업 중인 서버 인스턴스에서 채팅 메시지가 전송될 때 http://localhost:3459를 열 때 표시된 것처럼 해당 특정 서버에 연결된 클라이언트에게만 메시지가 로컬로 배포되는 방식으로 작동합니다. / 두 개의 다른 브라우저에서. 이제 서로 다른 두 서버를 통합하여 서로 통신할 수 있는 방법을 살펴보겠습니다.
4단계 - 메시지 브로커로서의 Redis
Redis는 빠르고 유연한 인메모리 데이터 구조 저장소입니다. 데이터를 캐시하기 위한 데이터베이스나 캐시 서버로 사용되는 경우가 많습니다. 또한 중앙 집중식 Pub/Sub 메시지 교환 패턴을 구현하는 데 사용할 수도 있습니다. Redis의 속도와 유연성으로 인해 분산 시스템에서 데이터를 공유하는 데 매우 인기 있는 선택이 되었습니다.
여기서 목표는 Redis를 메시지 브로커로 사용하여 채팅 서버를 통합하는 것입니다. 각 서버 인스턴스는 클라이언트(브라우저)로부터 받은 메시지를 동시에 메시지 브로커에 게시합니다. 메시지 브로커는 서버 인스턴스에서 오는 모든 메시지를 구독합니다.
app.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}`);
});
여기서는 Redis의 게시/구독 기능을 활용하고 있습니다. 두 개의 서로 다른 연결 인스턴스가 인스턴스화되었습니다. 하나는 메시지 게시용이고 다른 하나는 채널 구독용입니다. 클라이언트에서 메시지가 전송되면 redisPublisher 인스턴스의 publisher 메서드를 사용하여 “chat_messages”라는 Redis 채널에 메시지를 게시합니다. subscribe 메소드는 동일한 chat_message 채널을 구독하기 위해 redisSubscribe 인스턴스에서 호출됩니다. 이 채널에 메시지가 게시될 때마다 redisSubscriber.on 이벤트 리스너가 트리거됩니다. 이 이벤트 리스너는 현재 연결된 모든 WebSocket 클라이언트를 반복하고 수신된 메시지를 각 클라이언트에 보냅니다. 이는 한 사용자가 메시지를 보낼 때 서버 인스턴스에 연결된 다른 모든 사용자가 해당 메시지를 실시간으로 받도록 하기 위한 것입니다.
두 개의 서로 다른 서버를 시작하는 경우 다음과 같이 말합니다.
node app.js 3459
node app.js 3460
한 인스턴스에서 채팅 텍스트가 전송되면 이제 하나의 특정 서버가 아닌 연결된 서버 전체에 메시지를 브로드캐스트할 수 있습니다. http://localhost:3459/ 및 http://localhost:3460/를 실행한 다음 둘 사이에 채팅을 보내고 메시지가 브로드캐스트되는지 확인하여 이를 테스트할 수 있습니다. 실시간으로 두 서버에 걸쳐.
redis-cli에서 채널에 게시된 메시지를 모니터링하고 채널을 구독하여 구독된 메시지를 얻을 수도 있습니다.
redis-cli 명령을 실행합니다. 그런 다음 MONITOR를 입력하세요. 브라우저로 돌아가서 채팅을 시작하세요. 와우라는 채팅 텍스트를 보낸다고 가정하면 터미널에 다음과 같은 내용이 표시됩니다.
게시된 구독 메시지를 보려면 동일한 명령 redis-cli를 실행하고 SUBSCRIBE 채널 이름을 입력하세요. 우리의 경우 channelName은 chat_messages입니다. 브라우저에서 Great라는 텍스트를 보내면 터미널에 다음과 같은 내용이 표시되어야 합니다.
이제 서로 다른 포트 또는 심지어 서로 다른 시스템에서 서버의 여러 인스턴스를 실행할 수 있으며, 동일한 Redis 채널을 구독하는 한 연결된 모든 클라이언트에 메시지를 수신하고 브로드캐스트할 수 있으므로 사용자가 인스턴스 간에 원활하게 채팅할 수 있습니다.
소개 섹션에서 메시지 브로커를 사용한 Pub/Sub 패턴 구현에 대해 논의한 것을 기억하세요. 이 예는 이를 완벽하게 요약합니다.
위 그림에는 채팅 서버에 연결된 두 개의 서로 다른 클라이언트가 있습니다. 채팅 서버는 직접 연결되지 않고 Redis 인스턴스를 통해 상호 연결됩니다. 이는 클라이언트 연결을 독립적으로 처리하는 동시에 공통 매체(Redis)를 통해 정보(채팅 메시지)를 공유한다는 의미입니다. 위의 각 채팅 서버는 Redis에 연결됩니다. 이 연결은 Redis에 메시지를 게시하고 메시지를 수신하기 위해 Redis 채널을 구독하는 데 사용됩니다. 사용자가 메시지를 보내면 채팅 서버는 이를 Redis의 지정된 채널에 게시합니다.
Redis는 게시된 메시지를 받으면 이 메시지를 구독한 모든 채팅 서버에 브로드캐스트합니다. 그런 다음 각 채팅 서버는 연결된 모든 클라이언트에 메시지를 전달하여 연결된 서버에 관계없이 모든 사용자가 모든 사용자가 보낸 메시지를 받도록 보장합니다.
이 아키텍처를 사용하면 필요에 따라 더 많은 서버 인스턴스를 추가하여 채팅 애플리케이션을 수평적으로 확장할 수 있습니다. 각 인스턴스는 모든 인스턴스에서 일관된 메시지 배포를 보장하는 Redis의 게시/구독 시스템 기능 덕분에 자체적으로 연결된 클라이언트 세트를 처리할 수 있습니다. 이 설정은 많은 수의 동시 사용자를 처리하는 데 효율적이며 애플리케이션의 고가용성을 보장합니다.
결론
이 튜토리얼에서는 Redis를 메시지 브로커로 사용하여 이 패턴을 보여주기 위해 간단한 채팅 애플리케이션을 만드는 동안 게시/구독 패턴에 대해 배웠습니다. 다음은 단일 실패 지점(브로커 )이 있는 복잡한 분산 시스템과 같이 메시지 브로커가 최선의 솔루션이 아닐 수 있는 경우 P2P 메시징 시스템을 구현하는 방법을 배우는 것입니다. )는 선택사항이 아닙니다.
여기 GitHub에서 이 튜토리얼의 전체 소스 코드를 찾을 수 있습니다.
저자는 Write for DOnations 프로그램의 일환으로 No Kid Hungry를 기부 대상으로 선택했습니다.