Вступ
Патерн Pub/Sub - це універсальний односторонній патерн обміну повідомленнями, в якому паблішер генерує дані/повідомлення, а підписник реєструється для отримання певних типів повідомлень. Він може бути реалізований за допомогою однорангової архітектури або брокера повідомлень як посередника у передачі даних.
Наведене вище зображення ілюструє модель Peer-to-Peer Pub/Sub, де паблішер надсилає повідомлення безпосередньо підписникам без посередника. Підписникам потрібно знати адресу або кінцеву точку паблішера, щоб отримати повідомлення.
Примітка: Вузол у цьому випадку зазвичай означає активного учасника мережі обміну повідомленнями, яким може бути як служба, що публікує інформацію, так і служба, що отримує інформацію (абонент).
На наведеному вище зображенні модель Pub/Sub використовує брокер повідомлень як центральний вузол для обміну повідомленнями між паблішерами та підписниками. Брокер виступає посередником в обміні повідомленнями, розподіляючи повідомлення від паблішерів до підписників. Вузли підписуються на брокера, а не на паблішера напряму.
Присутність брокера покращує розділення між вузлами системи, оскільки і видавець, і підписники взаємодіють лише з брокером.
У цьому посібнику ви створите застосунок для чату в реальному часі для демонстрації цього патерну.
Попередні умови
- Node.js (версія >= 12) встановлений у вашій операційній системі.
- На вашому комп'ютері має бути встановлений редактор коду на кшталт VSCode.
- 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}`);
});
Для налаштування сервера буде використано метод createServer зі вбудованого модуля http в Node.js. Було задано PORT, на якому сервер повинен слухати запити, і викликано метод listen для прослуховування вхідних запитів на зазначеному порту.
У терміналі запустіть команду: node app.js, і у вас повинен з'явитися ось такий результат:
OutputServer is up and running on port 3459
Якщо ви відкриєте в браузері localhost: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>
Це базовий html-файл, який відображає Hello. Тепер ми повинні зчитати цей файл і відправити його як відповідь щоразу, коли на наш сервер надходить 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);
});
});
Тут ми використовуємо вбудований модуль path і функцію join для об'єднання сегментів. Потім за допомогою функції 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-сервера. Це дозволить нам обробляти як стандартні HTTP-запити, так і WebSocket-з'єднання через той самий порт 3459.
Подія з'єднання on() спрацьовує, коли встановлюється успішне з'єднання WebSocket. client у функції зворотного виклику - це об'єкт з'єднання 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, який ми налаштували спочатку, ми повинні створити екземпляр об'єкта WebSocket, вказавши ws://url, що ідентифікує сервер, який ми хочемо використовувати. Змінні url і socket, коли вони будуть записані, матимуть URL з'єднання з портом, де наш сервер слухає порт 3459, і об'єкт WebSocket, відповідно.
Отже, коли ви введете запит до сервера у вашому браузері, ви побачите ось це:
Тепер удосконалимо наш скрипт так, щоб ми могли надсилати повідомлення з клієнта на сервер і отримувати повідомлення з сервера.
<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-адресу, яка надсилає запит на наш сервер з боку клієнта (браузера), і створюємо новий екземпляр об'єкта WebSocket з цією URL-адресою. Потім ми генеруємо подію на елементі форми, коли натискається кнопка SendMessage. Береться текст, введений користувачем в інтерфейсі, і викликається метод send на екземплярі сокета для відправлення повідомлення на сервер.
Примітка: Для того, щоб відправити повідомлення на сервер через з'єднання WebSocket, зазвичай викликається методsend( )об'єкта WebSocket, який очікує один аргумент - повідомлення, який може бутиArrayBuffer,Blob,stringабо типізований масив. Цей метод буферизує вказане повідомлення для передачі та повертає його перед відправленням повідомлення на сервер.
Подія onmessage, що викликається в об'єкті сокета, спрацьовує при отриманні повідомлення від сервера. Це використовується для оновлення користувацького інтерфейсу вхідного повідомлення. Параметр eventMessage у функції зворотного виклику містить дані (повідомлення), надіслані з сервера, але повертаються вони як Blob. Потім до даних Blob застосовується метод text( ), який повертає promise, а потім за допомогою 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 для публікації/підписки. Було створено два різних екземпляри з'єднання, один для публікації повідомлень, а інший для підписки на канал. Коли повідомлення надсилається з клієнта, ми публікуємо його в каналі Redis з назвою chat_messages за допомогою методу publisher на екземплярі redisPublisher. Метод subscribe викликається на екземплярі redisSubscribe для підписки на той самий канал chat_message. Щоразу, коли повідомлення публікується у цьому каналі, спрацьовує обробник подій redisSubscriber.on. Цей обробник подій перебирає всі підключені на цей час клієнти WebSocket і надсилає отримане повідомлення кожному клієнту. Це робиться для того, щоб гарантувати, що коли один користувач надсилає повідомлення, всі інші користувачі, підключені до будь-якого екземпляра сервера, отримують це повідомлення у реальному часі.
Якщо ви запускаєте два різних сервери, скажімо:
node app.js 3459
node app.js 3460
Коли текст чату надсилається в одному екземплярі, ми можемо транслювати повідомлення на всі наші підключені сервери, а не лише на один конкретний сервер. Ви можете перевірити це, запустивши http://localhost:3459/ і http://localhost:3460/, а потім надіславши між ними повідомлення в чаті і побачите, що повідомлення транслюються між двома серверами в реальному часі.
Ви можете відстежувати повідомлення, опубліковані на каналі, за допомогою redis-cli, а також підписатися на канал, щоб отримувати повідомлення за підпискою:
Виконайте команду redis-cli. Потім введіть MONITOR. Поверніться в браузер і запустіть чат. У вашому терміналі ви повинні побачити щось на зразок цього, за умови, що ви відправили в чат текст Wow:
Щоб побачити опубліковані повідомлення, виконайте ту ж команду redis-cli і введіть SUBSCRIBE channelName. channelName в нашому випадку буде chat_messages. Щось подібне має бути у вашому терміналі, якщо ви надсилаєте повідомлення: Great з браузера:
Тепер ми можемо мати декілька екземплярів нашого сервера, що працюють на різних портах або навіть на різних машинах, і поки вони підписані на один канал Redis, вони можуть отримувати і транслювати повідомлення всім підключеним клієнтам, забезпечуючи користувачам можливість безперешкодного спілкування між різними серверами.
Пам'ятаєте, ми обговорювали реалізацію шаблону Pub/Sub за допомогою брокера повідомлень у вступному розділі. Цей приклад чудово підсумовує його.
На рисунку вище показано два різні клієнти, підключені до серверів чату. Сервери чату з'єднані між собою не напряму, а через екземпляр Redis. Це означає, що хоча вони обробляють клієнтські з'єднання незалежно, вони обмінюються інформацією (повідомленнями чату) через спільне середовище (Redis). Кожен сервер чату підключається до Redis. Це з'єднання використовується для публікації повідомлень в Redis і підписки на канали Redis для отримання повідомлень. Коли користувач надсилає повідомлення, чат-сервер публікує його у вказаному каналі Redis.
Коли Redis отримує опубліковане повідомлення, він транслює його всім підписаним на нього чат-серверам. Потім кожен чат-сервер ретранслює повідомлення всім підключеним клієнтам, гарантуючи, що кожен користувач отримає повідомлення, надіслані будь-яким користувачем, незалежно від того, до якого сервера він підключений.
Така архітектура дозволяє нам горизонтально масштабувати наш чат за допомогою додавання нових серверів за потреби. Кожен екземпляр може обробляти свій власний набір підключених клієнтів завдяки можливостям системи публікації/підписки Redis, яка забезпечує узгоджене розповсюдження повідомлень між усіма екземплярами. Таке налаштування є ефективним для роботи з великою кількістю одночасних користувачів і гарантує високу доступність вашого застосунку.
Висновок
У цьому уроці ми ознайомилися з патерном "Публікація/Підписка", створивши простий застосунок чату для демонстрації цього патерну, використовуючи Redis у якості брокера повідомлень.
Ви знайдете повний вихідний код цього підручника тут, на GitHub.
Ще немає коментарів