WebSocket в Python: реализация двусторонней связи в реальном времени

pip install websockets

import asyncio
import websockets

async def echo(websocket, path):
async for message in websocket:
print(f"Получено сообщение: {message}")
await websocket.send(f"Эхо: {message}")

async def main():
server = await websockets.serve(echo, "localhost", 8765)
print("Сервер запущен на ws://localhost:8765")
await server.wait_closed()

if __name__ == "__main__":
asyncio.run(main())

import asyncio
import json
import websockets

# Хранилище активных соединений
connected = set()

async def chat(websocket, path):
# Регистрируем нового клиента
connected.add(websocket)
try:
# Приветственное сообщение
await websocket.send(json.dumps({"type": "system", "message": "Добро пожаловать в чат!"}))

# Оповещаем остальных о новом участнике
if len(connected) > 1:
await notify_users(f"Новый пользователь присоединился к чату. Всего участников: {len(connected)}", websocket)

# Обрабатываем сообщения от клиента
async for message in websocket:
data = json.loads(message)
if data["type"] == "message":
await broadcast(data["message"], websocket)
except websockets.exceptions.ConnectionClosed:
print("Соединение закрыто клиентом")
finally:
# Удаляем клиента из списка при отключении
connected.remove(websocket)
if connected: # Если остались другие клиенты
await notify_users(f"Пользователь вышел из чата. Осталось участников: {len(connected)}")

async def notify_users(message, exclude=None):
"""Отправляет системное сообщение всем клиентам"""
system_message = json.dumps({"type": "system", "message": message})
if connected:
await asyncio.gather(
*[ws.send(system_message) for ws in connected if ws != exclude]
)

async def broadcast(message, sender):
"""Рассылает сообщение от одного клиента всем остальным"""
broadcast_message = json.dumps({"type": "chat", "message": message})
if connected:
await asyncio.gather(
*[ws.send(broadcast_message) for ws in connected if ws != sender]
)

async def main():
server = await websockets.serve(chat, "localhost", 8765)
print("Чат-сервер запущен на ws://localhost:8765")
await server.wait_closed()

if __name__ == "__main__":
asyncio.run(main())

import asyncio
import json
import websockets

# Простая имитация базы данных с токенами
USERS = {
"token1": {"username": "user1", "role": "admin"},
"token2": {"username": "user2", "role": "user"}
}

async def authenticate(websocket, path):
# Ждем первое сообщение с токеном
auth_message = await websocket.recv()
try:
data = json.loads(auth_message)
if "token" in data:
token = data["token"]
if token in USERS:
# Аутентификация успешна
user_info = USERS[token]
await websocket.send(json.dumps({
"type": "auth_success",
"username": user_info["username"]
}))

# Запускаем основную логику для аутентифицированного пользователя
await handle_messages(websocket, user_info)
return
except json.JSONDecodeError:
pass

# Если код дошел до сюда, аутентификация не удалась
await websocket.send(json.dumps({
"type": "auth_error",
"message": "Неверный токен аутентификации"
}))
await websocket.close(1008, "authentication failed")

async def handle_messages(websocket, user_info):
try:
async for message in websocket:
# Здесь логика обработки сообщений от аутентифицированного пользователя
print(f"Сообщение от {user_info['username']}: {message}")
await websocket.send(json.dumps({
"type": "response",
"message": f"Вы отправили: {message}"
}))
except websockets.exceptions.ConnectionClosed:
print(f"Соединение с {user_info['username']} закрыто")

async def main():
server = await websockets.serve(authenticate, "localhost", 8765)
print("Защищенный сервер запущен на ws://localhost:8765")
await server.wait_closed()

if __name__ == "__main__":
asyncio.run(main())

import asyncio
import websockets

async def send_and_receive():
uri = "ws://localhost:8765"
async with websockets.connect(uri) as websocket:
# Отправляем сообщение
message = "Привет, WebSocket!"
await websocket.send(message)
print(f"Отправлено: {message}")

# Получаем ответ
response = await websocket.recv()
print(f"Получено: {response}")

async def main():
# Запускаем клиент
await send_and_receive()

if __name__ == "__main__":
asyncio.run(main())

import asyncio
import websockets
import json
import sys

async def receive_messages(websocket):
"""Асинхронно получает сообщения от сервера"""
try:
while True:
message = await websocket.recv()
try:
data = json.loads(message)
if data["type"] == "chat":
print(f"\nНовое сообщение: {data['message']}")
elif data["type"] == "system":
print(f"\nСистема: {data['message']}")
else:
print(f"\nПолучено: {message}")
except json.JSONDecodeError:
print(f"\nПолучено: {message}")

# Повторно отображаем приглашение ввода
print("> ", end="", flush=True)
except websockets.exceptions.ConnectionClosed:
print("\nСоединение с сервером закрыто")
except Exception as e:
print(f"\nОшибка при получении сообщений: {e}")

async def send_messages(websocket):
"""Асинхронно отправляет сообщения на сервер"""
try:
# Отправляем токен для аутентификации
await websocket.send(json.dumps({"token": "token1"}))

while True:
print("> ", end="", flush=True)
message = await asyncio.get_event_loop().run_in_executor(None, sys.stdin.readline)
message = message.strip()

if message.lower() == "exit":
print("Завершение работы...")
break

await websocket.send(json.dumps({"type": "message", "message": message}))
except websockets.exceptions.ConnectionClosed:
print("\nСоединение с сервером закрыто")
except Exception as e:
print(f"\nОшибка при отправке сообщения: {e}")

async def main():
uri = "ws://localhost:8765"
try:
async with websockets.connect(uri) as websocket:
print(f"Подключено к {uri}")
print("Введите сообщения для отправки. Для выхода введите 'exit'")

# Запускаем обе задачи параллельно
receiver_task = asyncio.create_task(receive_messages(websocket))
sender_task = asyncio.create_task(send_messages(websocket))

# Ждем завершения любой из задач
done, pending = await asyncio.wait(
[receiver_task, sender_task],
return_when=asyncio.FIRST_COMPLETED
)

# Отменяем оставшуюся задачу
for task in pending:
task.cancel()
except websockets.exceptions.ConnectionError:
print(f"Не удалось подключиться к {uri}")
except Exception as e:
print(f"Ошибка: {e}")

if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\nКлиент остановлен пользователем")

import asyncio
import websockets
import ssl
import pathlib

async def secure_client():
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
localhost_pem = pathlib.Path(__file__).with_name("localhost.pem")
ssl_context.load_verify_locations(localhost_pem)

uri = "wss://localhost:8765"
async with websockets.connect(uri, ssl=ssl_context) as websocket:
await websocket.send("Безопасное соединение установлено!")
response = await websocket.recv()
print(f"Получено: {response}")

async def main():
await secure_client()

if __name__ == "__main__":
asyncio.run(main())

from flask import Flask, render_template
from flask_socketio import SocketIO, emit, join_room, leave_room

app = Flask(__name__)
app.config['SECRET_KEY'] = 'секретный-ключ'
socketio = SocketIO(app, cors_allowed_origins="*")

# Словарь для хранения пользователей и их комнат
users = {}

@app.route('/')
def index():
return render_template('index.html')

@socketio.on('connect')
def handle_connect():
print('Клиент подключился')

@socketio.on('disconnect')
def handle_disconnect():
if request.sid in users:
room = users[request.sid]["room"]
username = users[request.sid]["username"]
leave_room(room)
del users[request.sid]
emit('status', {'msg': f'{username} покинул чат'}, room=room)

@socketio.on('join')
def handle_join(data):
username = data['username']
room = data['room']
join_room(room)
users[request.sid] = {"username": username, "room": room}
emit('status', {'msg': f'{username} присоединился к чату'}, room=room)

@socketio.on('message')
def handle_message(data):
room = users[request.sid]["room"]
username = users[request.sid]["username"]
emit('message', {'msg': data['msg'], 'username': username}, room=room)

if __name__ == '__main__':
socketio.run(app, debug=True)

<!-- index.html -->
<!DOCTYPE html>
<html>
<head>
<title>Flask-SocketIO Чат</title>
<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script>
<script>
document.addEventListener('DOMContentLoaded', function() {
const socket = io();

// Присоединение к комнате
document.getElementById('join-form').addEventListener('submit', function(e) {
e.preventDefault();
const username = document.getElementById('username').value;
const room = document.getElementById('room').value;
socket.emit('join', {username: username, room: room});

document.getElementById('join-container').style.display = 'none';
document.getElementById('chat-container').style.display = 'block';
});

// Отправка сообщения
document.getElementById('message-form').addEventListener('submit', function(e) {
e.preventDefault();
const messageInput = document.getElementById('message');
socket.emit('message', {msg: messageInput.value});
messageInput.value = '';
});

// Получение сообщений
socket.on('message', function(data) {
addMessage(`<strong>${data.username}</strong>: ${data.msg}`);
});

// Получение статусных сообщений
socket.on('status', function(data) {
addMessage(`<i>${data.msg}</i>`);
});

function addMessage(html) {
const messageDiv = document.createElement('div');
messageDiv.innerHTML = html;
document.getElementById('messages').appendChild(messageDiv);
// Прокрутка вниз
document.getElementById('messages').scrollTop = document.getElementById('messages').scrollHeight;
}
});
</script>
</head>
<body>
<div id="join-container">
<h2>Присоединиться к чату</h2>
<form id="join-form">
<input id="username" placeholder="Ваше имя" required>
<input id="room" placeholder="Название комнаты" required>
<button type="submit">Присоединиться</button>
</form>
</div>

<div id="chat-container" style="display: none;">
<h2>Чат</h2>
<div id="messages" style="height: 300px; overflow-y: scroll; border: 1px solid #ccc;"></div>
<form id="message-form">
<input id="message" placeholder="Введите сообщение" autocomplete="off" required>
<button type="submit">Отправить</button>
</form>
</div>
</body>
</html>

# server.py
import asyncio
import json
import psutil
import websockets
import datetime

# Хранилище активных клиентских соединений
connected = set()

async def send_system_stats():
"""Отправляет статистику всем подключенным клиентам"""
while True:
if connected: # Проверяем, есть ли подключенные клиенты
# Собираем статистику
stats = {
'timestamp': datetime.datetime.now().strftime('%H:%M:%S'),
'cpu_percent': psutil.cpu_percent(interval=None),
'memory_percent': psutil.virtual_memory().percent,
'disk_percent': psutil.disk_usage('/').percent,
'network': {
'bytes_sent': psutil.net_io_counters().bytes_sent,
'bytes_recv': psutil.net_io_counters().bytes_recv
}
}

# Преобразуем в JSON
message = json.dumps(stats)

# Отправляем всем клиентам
websockets_to_remove = set()
for websocket in connected:
try:
await websocket.send(message)
except websockets.exceptions.ConnectionClosed:
# Отмечаем закрытые соединения для последующего удаления
websockets_to_remove.add(websocket)

# Удаляем закрытые соединения
connected -= websockets_to_remove

# Ждем 1 секунду перед следующим обновлением
await asyncio.sleep(1)

async def handler(websocket, path):
"""Обработчик подключений"""
# Добавляем новое соединение в множество
connected.add(websocket)
print(f"Новое подключение: {len(connected)} активных клиентов")

try:
# Отправляем приветственное сообщение
await websocket.send(json.dumps({
'type': 'init',
'message': 'Соединение установлено. Начинаем передачу данных.'
}))

# Просто держим соединение открытым
await websocket.wait_closed()
finally:
# Удаляем соединение при отключении
connected.remove(websocket)
print(f"Клиент отключился: {len(connected)} активных клиентов")

async def main():
# Запускаем сервер
server = await websockets.serve(handler, "0.0.0.0", 8765)
print("Сервер мониторинга запущен на ws://0.0.0.0:8765")

# Запускаем задачу отправки статистики
asyncio.create_task(send_system_stats())

# Держим сервер запущенным
await server.wait_closed()

if __name__ == "__main__":
asyncio.run(main())

class WebSocketConnection:
def __init__(self, uri):
self.uri = uri
self.websocket = None

async def __aenter__(self):
try:
self.websocket = await websockets.connect(self.uri)
return self.websocket
except Exception as e:
print(f"Ошибка подключения: {e}")
raise

async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.websocket:
await self.websocket.close()

if exc_type is not None:
print(f"Произошла ошибка: {exc_val}")
# Возвращаем False, чтобы исключение было поднято дальше
return False

async def connect_with_retry(uri, max_retries=5, backoff_factor=1.5):
"""Подключается к WebSocket с автоматическими повторными попытками"""
retry_count = 0
delay = 1 # начальная задержка в секундах

while retry_count < max_retries:
try:
return await websockets.connect(uri)
except (websockets.exceptions.ConnectionClosed, 
websockets.exceptions.InvalidStatusCode,
ConnectionRefusedError) as e:
retry_count += 1
if retry_count >= max_retries:
raise Exception(f"Не удалось подключиться после {max_retries} попыток: {e}")

print(f"Попытка {retry_count}/{max_retries} не удалась: {e}")
print(f"Повторная попытка через {delay} секунд...")
await asyncio.sleep(delay)
delay *= backoff_factor # увеличиваем задержку с каждой попыткой

async def heartbeat(websocket, interval=30):
"""Отправляет ping-сообщения для поддержания соединения"""
try:
while True:
await asyncio.sleep(interval)
try:
pong = await websocket.ping()
await asyncio.wait_for(pong, timeout=10)
print("Получен pong от сервера")
except asyncio.TimeoutError:
raise websockets.exceptions.ConnectionClosed(
1011, "Сервер не ответил на ping"
)
except websockets.exceptions.ConnectionClosed:
print("Соединение закрыто во время heartbeat")
# Здесь можно добавить логику переподключения

import asyncio
import zlib
import json

# Ограничиваем количество одновременных соединений
connection_semaphore = asyncio.Semaphore(100)

# Буфер для сообщений
message_buffer = {}
BUFFER_SIZE = 10

async def send_with_compression(websocket, message):
"""Отправляет сжатые данные через WebSocket"""
json_data = json.dumps(message).encode('utf-8')
compressed_data = zlib.compress(json_data)
await websocket.send(compressed_data)

async def buffer_message(websocket, message):
"""Буферизует сообщения и отправляет их пакетами"""
client_id = id(websocket)

if client_id not in message_buffer:
message_buffer[client_id] = []

message_buffer[client_id].append(message)

# Если буфер полон, отправляем пакет сообщений
if len(message_buffer[client_id]) >= BUFFER_SIZE:
await send_messages_batch(websocket, client_id)

async def send_messages_batch(websocket, client_id):
"""Отправляет накопленные сообщения одним пакетом"""
if client_id in message_buffer and message_buffer[client_id]:
messages = message_buffer[client_id]
await send_with_compression(websocket, {"type": "batch", "messages": messages})
message_buffer[client_id] = []

import asyncio
import websockets
import redis.asyncio as redis
import json

# Подключение к Redis
redis_client = None

async def setup_redis():
global redis_client
redis_client = await redis.Redis(host='localhost', port=6379, db=0)

# Подписка на канал для получения сообщений от других экземпляров сервера
pubsub = redis_client.pubsub()
await pubsub.subscribe("websocket_broadcast")

# Задача для прослушивания сообщений
asyncio.create_task(listen_redis_messages(pubsub))

async def listen_redis_messages(pubsub):
"""Слушает сообщения от Redis и рассылает их подключенным клиентам"""
while True:
message = await pubsub.get_message(ignore_subscribe_messages=True)
if message:
data = json.loads(message['data'].decode('utf-8'))
await broadcast_to_clients(data)

async def broadcast_to_clients(message, exclude_websocket=None):
"""Отправляет сообщение всем подключенным клиентам"""
# Логика отправки сообщений клиентам
pass

async def broadcast_via_redis(message):
"""Отправляет сообщение через Redis для всех экземпляров сервера"""
if redis_client:
await redis_client.publish("websocket_broadcast", json.dumps(message))

import time
import functools
import asyncio

def async_timeit(func):
@functools.wraps(func)
async def async_wrapper(*args, **kwargs):
start_time = time.time()
result = await func(*args, **kwargs)
end_time = time.time()
print(f"{func.__name__} выполнилась за {end_time – start_time:.4f} секунд")
return result
return async_wrapper

@async_timeit
async def process_message(message):
"""Обрабатывает сообщение с измерением времени выполнения"""
await asyncio.sleep(0.1) # Имитация работы
return {"status": "processed"}