Введение
Паттерн «издатель/подписчик» — это универсальный односторонний механизм обмена сообщениями, в котором издатель генерирует данные/сообщения, а подписчики подписываются на получение сообщений определённых типов. Он может быть реализован с использованием одноранговой архитектуры или брокера сообщений для посредничества в обмене данными.
На изображении выше показана модель одноранговой сети «издатель-подписчик», в которой издатель отправляет сообщения напрямую подписчикам без посредников. Подписчикам необходимо знать адрес или конечную точку издателя, чтобы получить сообщение.
На изображении выше модель «издатель/подписчик» использует брокера сообщений в качестве центрального узла для отправки сообщений между издателями и подписчиками. Брокер выступает посредником в обмене сообщениями и распределяет сообщения от издателей к подписчикам. Узлы-подписчики подписываются непосредственно на брокера, а не на издателя. Наличие брокера улучшает изоляцию узлов системы, поскольку и издатель, и подписчики взаимодействуют только с брокером. В этом руководстве вы создадите приложение для чата в реальном времени, чтобы дополнительно продемонстрировать эту модель.
Предпосылки
- 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}`);
});Метод createServer В модуле http Одомашненный Node.js Он будет использоваться для запуска сервера. Задается порт, на котором сервер должен принимать запросы, и вызывается метод listen созданного экземпляра сервера для прослушивания входящих запросов на указанном порту.
Заказ node app.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>Это простой HTML-файл, который отображает сообщение "Hello". Теперь нам нужно прочитать этот файл и отобразить его в качестве ответа при каждом 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);
});
});Здесь мы используем внутренний модуль пути и функцию. присоединиться Мы используем это для соединения сегментов пути. Затем мы используем эту функцию. readFile Чтобы прочитать файл index.html Используется асинхронно. Требует два аргумента: путь к файлу для чтения и результат чтения. Код состояния. 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-сервер под названием... webSocketServer Мы создаём его и отправляем на сервер. HTTP Мы подключаемся к существующему порту. Это позволяет нам обрабатывать как стандартные HTTP-запросы, так и WebSocket-соединения на одном порту 3459.
Событие on() срабатывает при успешном установлении соединения WebSocket. Клиент внутри функции перезвонить Объект соединения WebSocket представляет собой соединение с клиентом. Он используется для отправки и получения сообщений, а также для прослушивания событий, таких как сообщения от клиента.
Функция distributeClientMessages используется здесь для отправки входящих сообщений всем подключенным клиентам. Она принимает аргумент 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}`);
</script>
</body>
</html>В этом фрагменте кода мы добавили элемент формы, содержащий поле ввода и кнопку для отправки сообщения. Соединения WebSocket инициируются клиентами, и для связи с сервером WebSocket, который мы изначально настроили, нам необходимо создать экземпляр объекта WebSocket, указывающий ws://url, то есть сервер, который мы хотим использовать. Переменная URL при входе в систему будет содержать URL-адрес, подключающийся к порту 3459, где наш сервер прослушивает соединения.
// 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-адрес, который отправляет запрос на наш сервер от клиента (браузера), и создаем новый экземпляр объекта WebSocket с этим URL-адресом. Затем, нажав кнопку, мы получаем URL-адрес. Отправить сообщение Мы создаём событие для элемента формы. Текст, введённый пользователем в пользовательском интерфейсе, извлекается из элемента ввода, и вызывается метод send экземпляра сокета для отправки сообщения на сервер.
Событие сообщение Вызывается у объекта сокета и срабатывает при получении сообщения от сервера. Используется для обновления пользовательского интерфейса полученного сообщения. Параметр eventMessage В функции перезвонить В нем содержатся данные (сообщение), отправленные с сервера, но возвращенные в виде объекта Blob. Затем к данным Blob применяется метод text(), который возвращает промис, и он разрешается с помощью метода 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 экземпляра redisPublisher. ""chat_messages"" Мы публикуем сообщение. Вызывается метод 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 Запустите. Затем введите МОНИТОР Вернитесь в браузер и начните общаться в чате. В терминале вы должны увидеть что-то подобное, если отправите сообщение в чате WoW:
Для просмотра опубликованных сообщений, которыми поделились другие пользователи, используйте ту же команду. redis-cli Беги и ПОДПИСАТЬСЯ на канал Введите название канала и информацию о нас. сообщения в чате Если вы отправите текстовое сообщение, в терминале должно отобразиться что-то подобное: Отлично работает в браузере:
Теперь мы можем запускать несколько экземпляров нашего сервера на разных портах или даже на разных машинах, и пока они подписаны на один и тот же канал Redis, они могут получать и передавать сообщения всем подключенным клиентам, обеспечивая бесперебойное общение пользователей между экземплярами.
Помните, во вводной части мы говорили о реализации шаблона Pub/Sub с использованием брокера сообщений? Этот пример прекрасно это иллюстрирует.
На приведенном выше рисунке к серверам чата подключены 2 разных клиента (браузера). Серверы чата не соединены напрямую, а через экземпляр Redis. Это означает, что, хотя они обрабатывают клиентские соединения независимо, они обмениваются информацией (сообщениями чата) по общей среде (Redis). Каждый из серверов чата подключается к Redis. Это соединение используется для публикации сообщений в Redis и подписки на каналы Redis для получения сообщений. Когда пользователь отправляет сообщение, сервер чата публикует его в указанный канал в Redis.
Когда Redis получает опубликованное сообщение, он рассылает его всем участвующим чат-серверам. Затем каждый чат-сервер отправляет сообщение всем подключенным клиентам, гарантируя, что каждый пользователь получит сообщения, отправленные каждым пользователем, независимо от того, к какому серверу он подключен.
Эта архитектура позволяет масштабировать наше приложение для чата горизонтально, добавляя необходимые экземпляры сервера. Благодаря возможностям системы публикации/подписки Redis, обеспечивающей равномерное распределение сообщений по всем экземплярам, каждый экземпляр может управлять своим собственным набором подключенных клиентов. Такая конфигурация эффективна для обработки большого количества одновременно работающих пользователей и гарантирует доступность вашего приложения.
Результат
В этом уроке мы изучили шаблон «публикация/подписка», создав простое приложение для чата, чтобы продемонстрировать этот шаблон, используя Redis в качестве брокера сообщений. Следующий шаг — изучить, как реализовать одноранговую систему обмена сообщениями в случаях, когда брокер сообщений может быть не лучшим решением, например, в сложных распределенных системах, где единая точка отказа (брокер) недопустима.
