Как создать веб-приложения реального времени с Node.js и WebSocket

const WebSocket = require('ws');

// Создаем сервер WebSocket на порту 8080
const wss = new WebSocket.Server({ port: 8080 });

// Обработчик подключения нового клиента
wss.on('connection', function connection(ws) {
console.log('Новый клиент подключен');

// Обработка входящих сообщений
ws.on('message', function incoming(message) {
console.log('Получено: %s', message);

// Отправляем сообщение обратно клиенту
ws.send(`Эхо: ${message}`);
});

// Отправляем приветственное сообщение
ws.send('Добро пожаловать на сервер!');
});

console.log('WebSocket-сервер запущен на порту 8080');

const express = require('express');
const http = require('http');
const { Server } = require('socket.io');

const app = express();
const server = http.createServer(app);
const io = new Server(server);

// Раздаем статические файлы
app.use(express.static('public'));

// Обработка WebSocket-подключений
io.on('connection', (socket) => {
console.log(`Пользователь ${socket.id} подключился`);

// Обработка события от клиента
socket.on('chat message', (msg) => {
console.log(`Сообщение: ${msg}`);
// Рассылаем сообщение всем клиентам
io.emit('chat message', msg);
});

// Обработка отключения
socket.on('disconnect', () => {
console.log(`Пользователь ${socket.id} отключился`);
});
});

server.listen(3000, () => {
console.log('Сервер запущен на порту 3000');
});

// С использованием socket.io
io.use((socket, next) => {
// Получаем токен из запроса
const token = socket.handshake.auth.token;

// Проверяем токен
verifyToken(token)
.then(user => {
// Сохраняем данные пользователя в объекте сокета
socket.user = user;
next();
})
.catch(err => {
next(new Error('Неверный токен аутентификации'));
});
});

wss.on('error', (error) => {
console.error('WebSocket Server Error:', error);
// Здесь можно добавить отправку уведомлений команде
});

// Мониторинг состояния соединений
setInterval(() => {
wss.clients.forEach((client) => {
if (client.isAlive === false) return client.terminate();

client.isAlive = false;
client.ping(() => {});
});
}, 30000);

// Устанавливаем соединение с сервером
const socket = new WebSocket('ws://localhost:8080');

// Обработчик открытия соединения
socket.addEventListener('open', (event) => {
console.log('Соединение установлено');

// Отправляем сообщение серверу
socket.send('Привет, сервер!');
});

// Обработчик входящих сообщений
socket.addEventListener('message', (event) => {
console.log(`Получено сообщение: ${event.data}`);
});

// Обработчик ошибок
socket.addEventListener('error', (event) => {
console.error('WebSocket ошибка:', event);
});

// Обработчик закрытия соединения
socket.addEventListener('close', (event) => {
console.log(`Соединение закрыто, код: ${event.code}, причина: ${event.reason}`);
});

// Функция для отправки сообщений
function sendMessage(message) {
// Проверяем готовность соединения
if (socket.readyState === WebSocket.OPEN) {
socket.send(message);
} else {
console.warn('Соединение не готово, сообщение не отправлено');
}
}

// Подключаем библиотеку socket.io-client
// <script src="https://cdn.socket.io/4.4.1/socket.io.min.js"></script>

// Устанавливаем соединение
const socket = io('http://localhost:3000', {
auth: {
token: 'пользовательский_токен' // для аутентификации
}
});

// Обработчик подключения
socket.on('connect', () => {
console.log('Подключение установлено, ID:', socket.id);
});

// Подписываемся на события
socket.on('chat message', (msg) => {
console.log('Новое сообщение:', msg);
// Добавляем сообщение в чат
addMessageToChat(msg);
});

// Отправка сообщений
function sendChatMessage(message) {
socket.emit('chat message', message);
}

// Обработка ошибок
socket.on('connect_error', (error) => {
console.error('Ошибка подключения:', error.message);
// Показываем уведомление пользователю
showErrorNotification(error.message);
});

// Отключение при закрытии страницы
window.addEventListener('beforeunload', () => {
socket.disconnect();
});

// Отправка структурированных данных
socket.emit('user action', {
type: 'update_profile',
data: {
name: 'Иван Петров',
status: 'онлайн',
lastActive: new Date().toISOString()
},
timestamp: Date.now()
});

// Обработка на сервере
socket.on('user action', (data) => {
switch(data.type) {
case 'update_profile':
updateUserProfile(socket.id, data.data);
break;
case 'request_data':
sendDataToUser(socket, data.data.dataType);
break;
default:
console.warn('Неизвестный тип действия:', data.type);
}
});

// Функция для создания WebSocket с автоматическим переподключением
function createReconnectingWebSocket(url) {
let socket;
let reconnectAttempts = 0;
const maxReconnectAttempts = 5;
const reconnectInterval = 1000; // начальный интервал 1 секунда

const connect = () => {
socket = new WebSocket(url);

socket.addEventListener('open', () => {
console.log('Соединение установлено');
reconnectAttempts = 0; // сбрасываем счетчик при успешном подключении
});

socket.addEventListener('close', (event) => {
if (reconnectAttempts < maxReconnectAttempts) {
const timeout = reconnectInterval * Math.pow(2, reconnectAttempts);
console.log(`Переподключение через ${timeout}мс...`);
setTimeout(connect, timeout);
reconnectAttempts++;
} else {
console.error('Превышено количество попыток подключения');
}
});

// Другие обработчики событий
};

connect();
return {
send: (data) => {
if (socket && socket.readyState === WebSocket.OPEN) {
socket.send(data);
return true;
}
return false;
},
close: () => socket && socket.close()
};
}

// Использование
const ws = createReconnectingWebSocket('ws://localhost:8080');

const cluster = require('cluster');
const http = require('http');
const { Server } = require('socket.io');
const numCPUs = require('os').cpus().length;
const { createAdapter } = require('@socket.io/cluster-adapter');
const { setupMaster } = require('@socket.io/sticky');

if (cluster.isPrimary) {
console.log(`Основной процесс ${process.pid} запущен`);

// Создаем HTTP-сервер
const httpServer = http.createServer();

// Настраиваем sticky sessions
setupMaster(httpServer, {
loadBalancingMethod: "least-connection"
});

// Открываем порт
httpServer.listen(3000);

// Запускаем рабочие процессы
for (let i = 0; i < numCPUs; i++) {
cluster.fork();
}

cluster.on('exit', (worker, code, signal) => {
console.log(`Рабочий процесс ${worker.process.pid} завершился`);
// Перезапускаем упавший процесс
cluster.fork();
});
} else {
console.log(`Рабочий процесс ${process.pid} запущен`);

// Создаем HTTP-сервер
const httpServer = http.createServer();
const io = new Server(httpServer);

// Используем адаптер для обмена сообщениями между рабочими процессами
io.adapter(createAdapter());

// Обработка соединений
io.on('connection', (socket) => {
console.log(`Новое подключение ${socket.id} в процессе ${process.pid}`);
// ... обработка событий
});

// Не нужно запускать сервер на отдельном порту,
// это уже сделал основной процесс
}

// Подключение Redis-адаптера
const { createAdapter } = require('@socket.io/redis-adapter');
const { createClient } = require('redis');

const pubClient = createClient({ url: 'redis://localhost:6379' });
const subClient = pubClient.duplicate();

Promise.all([pubClient.connect(), subClient.connect()]).then(() => {
io.adapter(createAdapter(pubClient, subClient));
io.listen(3000);
});

// Увеличение лимита открытых файловых дескрипторов
process.setMaxListeners(0);
require('events').EventEmitter.defaultMaxListeners = 0;

// На уровне операционной системы также может потребоваться
// настройка параметров ядра (sysctl для Linux):
// sysctl -w fs.file-max=500000
// sysctl -w net.ipv4.ip_local_port_range="1024 65535"

// Серверная часть (с использованием Socket.io)
io.on('connection', (socket) => {
// Присоединение к комнате при входе в чат
socket.on('join_room', (roomId) => {
socket.join(roomId);
socket.to(roomId).emit('user_joined', socket.id);
});

// Обработка новых сообщений
socket.on('message', (data) => {
// Сохраняем сообщение в базе данных
saveMessageToDatabase(data)
.then(() => {
// Отправляем сообщение всем в комнате
io.to(data.roomId).emit('message', {
id: generateUniqueId(),
text: data.text,
sender: socket.id,
timestamp: Date.now()
});
});
});

// Индикатор печати
socket.on('typing', (roomId) => {
socket.to(roomId).emit('user_typing', socket.id);
});

// Обработка отключения
socket.on('disconnect', () => {
// Уведомляем все комнаты, где был пользователь
notifyRoomsAboutDisconnect(socket.id);
});
});

// Реализация простого совместного редактора
io.on('connection', (socket) => {
socket.on('join_document', async (documentId) => {
// Загружаем текущее состояние документа
const document = await getDocumentFromDatabase(documentId);

// Присоединяем к комнате документа
socket.join(`doc:${documentId}`);

// Отправляем текущее состояние
socket.emit('document_state', document);

// Обновляем список активных редакторов
io.to(`doc:${documentId}`).emit('editors_update', 
getActiveEditorsInRoom(`doc:${documentId}`));
});

// Обработка изменений в документе
socket.on('text_operation', (data) => {
// Применяем операционную трансформацию для 
// корректного слияния изменений
const transformedOp = transformOperation(data.operation);

// Обновляем документ в базе данных
updateDocumentWithOperation(data.documentId, transformedOp);

// Рассылаем операцию другим редакторам
socket.to(`doc:${data.documentId}`).emit('text_operation', {
operation: transformedOp,
author: socket.id
});
});

// Индикация позиции курсора
socket.on('cursor_position', (data) => {
socket.to(`doc:${data.documentId}`).emit('remote_cursor', {
position: data.position,
author: socket.id
});
});
});

// Серверная логика для простой многопользовательской игры
const games = new Map();

io.on('connection', (socket) => {
socket.on('join_game', (gameId) => {
// Создаем игру, если она не существует
if (!games.has(gameId)) {
games.set(gameId, {
players: [],
state: initGameState()
});
}

const game = games.get(gameId);

// Добавляем игрока
game.players.push({
id: socket.id,
position: getRandomStartPosition(),
score: 0
});

// Присоединяем к комнате игры
socket.join(gameId);

// Отправляем текущее состояние игры
socket.emit('game_state', game.state);

// Уведомляем других игроков
socket.to(gameId).emit('player_joined', {
id: socket.id,
position: game.players[game.players.length – 1].position
});
});

// Обработка движений игрока
socket.on('player_move', (data) => {
const game = games.get(data.gameId);
if (!game) return;

// Обновляем позицию игрока
const player = game.players.find(p => p.id === socket.id);
if (player) {
player.position = validateAndUpdatePosition(player.position, data.movement);

// Отправляем обновление всем игрокам
io.to(data.gameId).emit('player_position', {
id: socket.id,
position: player.position
});
}
});
});

// Игровой цикл для обновления состояния
setInterval(() => {
games.forEach((game, gameId) => {
// Обновляем игровую логику
updateGameLogic(game);

// Отправляем обновления
io.to(gameId).emit('game_update', getGameUpdateData(game));
});
}, 100); // 10 обновлений в секунду

// Трансляция рыночных данных
const marketDataStream = connectToMarketDataProvider();

marketDataStream.on('price_update', (data) => {
// Фильтруем и форматируем данные
const formattedData = formatMarketData(data);

// Отправляем всем подписанным клиентам
io.to(`market:${data.instrument}`).emit('price_update', formattedData);
});

io.on('connection', (socket) => {
// Аутентификация и авторизация
authenticateUser(socket)
.then(user => {
socket.user = user;

// Подписка на инструменты
socket.on('subscribe_instrument', (instrument) => {
// Проверяем права доступа
if (hasAccessToInstrument(socket.user, instrument)) {
socket.join(`market:${instrument}`);

// Отправляем последнюю известную цену
socket.emit('price_update', getLatestPrice(instrument));
} else {
socket.emit('error', { message: 'Недостаточно прав для доступа к инструменту' });
}
});

// Размещение ордера
socket.on('place_order', async (order) => {
try {
// Валидация ордера
validateOrder(order);

// Размещение в системе
const result = await placeOrderInTradingSystem(socket.user, order);

// Подтверждение клиенту
socket.emit('order_placed', result);
} catch (error) {
socket.emit('order_error', { message: error.message });
}
});
})
.catch(error => {
socket.emit('auth_error', { message: error.message });
socket.disconnect();
});
});

// Система мониторинга серверов
const monitoring = require('./monitoring');

// Запуск сборщиков метрик
monitoring.startCollectors();

io.on('connection', (socket) => {
// Аутентификация для доступа к панели мониторинга
socket.on('auth', (credentials) => {
authenticateAdmin(credentials)
.then(() => {
socket.isAuthenticated = true;
socket.emit('auth_success');
})
.catch(() => {
socket.emit('auth_failed');
});
});

// Запрос данных мониторинга
socket.on('get_metrics', (serverIds) => {
if (!socket.isAuthenticated) return;

// Получаем текущие метрики
const metrics = monitoring.getMetricsForServers(serverIds);
socket.emit('metrics_data', metrics);

// Подписка на обновления
serverIds.forEach(serverId => {
socket.join(`server:${serverId}`);
});
});

// Выполнение команды на сервере
socket.on('execute_command', (data) => {
if (!socket.isAuthenticated) return;

monitoring.executeCommandOnServer(data.serverId, data.command)
.then(result => {
socket.emit('command_result', {
serverId: data.serverId,
command: data.command,
result
});
})
.catch(error => {
socket.emit('command_error', {
serverId: data.serverId,
command: data.command,
error: error.message
});
});
});
});

// Публикация обновлений метрик
monitoring.on('metrics_update', (serverId, metrics) => {
io.to(`server:${serverId}`).emit('metrics_update', {
serverId,
metrics,
timestamp: Date.now()
});
});

// Оповещение о критических событиях
monitoring.on('alert', (alert) => {
io.to('admins').emit('server_alert', {
...alert,
timestamp: Date.now()
});
});