Асинхронные веб-сокеты в Python: как создать реалтайм-чат с websockets и asyncio

Представьте, что вы пишете чат, где сообщения приходят мгновенно - без перезагрузки страницы, без нажатия кнопки «Обновить». Или игру, где движения игроков синхронизируются в реальном времени. Или трейдинговую платформу, где цены обновляются каждую миллисекунду. Всё это возможно - и делается проще, чем вы думаете, с помощью асинхронных веб-сокетов в Python.

Раньше, чтобы получить обновление с сервера, клиент должен был постоянно спрашивать: «Есть что-нибудь новое?». Это называлось polling. Но когда сотни или тысячи пользователей одновременно спрашивают каждые 2 секунды, сервер начинает дышать тяжело. Веб-сокеты убирают этот лишний шум. Они создают постоянный канал связи: как телефонный разговор, а не SMS-переписка. И в Python, особенно с websockets и asyncio, это становится почти тривиально просто.

Что такое веб-сокет и почему он лучше HTTP

HTTP - это протокол, где клиент отправляет запрос, сервер отвечает, и соединение закрывается. Это надёжно, но медленно для постоянного обмена данными. Веб-сокеты работают иначе: сначала происходит «рукопожатие» - клиент отправляет специальный HTTP-запрос, сервер отвечает с подтверждением, и после этого соединение превращается в постоянный двунаправленный канал. Данные летят туда-сюда без лишних заголовков, без повторного подключения. Это как открыть трубу и просто лить воду - не нужно каждый раз открывать кран.

Задержка? Может быть меньше 10 миллисекунд. Трафик? На 90% меньше, чем при постоянных запросах. И главное - сервер может отправить данные клиенту, когда захочет, а не когда клиент его спросит. Это принципиально меняет архитектуру приложений.

Как работает асинхронность в Python

Python - не многопоточный язык по умолчанию. Он использует один поток, но может выполнять множество задач, пока одна из них ждёт. Это называется асинхронностью. Библиотека asyncio позволяет писать код, который не блокируется при ожидании сетевых операций. Вместо того чтобы стоять и ждать ответа от клиента, сервер переходит к другому соединению. Когда первый клиент готов - система возвращается к нему. Это как официант, который обслуживает сразу несколько столов, не стоя у одного и не ждёт, пока все съедят.

Когда вы пишете асинхронный веб-сокет, вы не пишете «пока ждём ответа, не делай ничего». Вы пишете: «Когда пришло сообщение - обработай его. Пока ждёшь - займись другим клиентом». Это позволяет одному серверу обслуживать десятки тысяч одновременных соединений на обычном VPS.

Простой echo-сервер на websockets

Начнём с самого базового примера - сервер, который просто возвращает всё, что ему присылают. Это как «Hello, World!» для веб-сокетов.

import asyncio
import websockets

async def echo(websocket):
    async for message in websocket:
        await websocket.send(message)

async def main():
    async with websockets.serve(echo, "localhost", 8765):
        await asyncio.Future()  # сервер будет работать до остановки

asyncio.run(main())

Всё. Это работает. Запустите этот код - и у вас есть полноценный WebSocket-сервер. Подключитесь через браузер (или любой WebSocket-клиент), отправьте текст - и он вернётся вам. Никаких HTTP-запросов. Никаких таймаутов. Просто сообщение → ответ.

Как это работает? Функция echo - это асинхронный генератор. Она ждёт, пока клиент отправит сообщение (async for), и сразу же отправляет его обратно. websockets.serve запускает сервер на порту 8765. asyncio.Future() просто удерживает цикл запущенным - как вечный цикл while True, но без нагрузки.

Как сделать настоящий чат

Echo - это забавно, но не очень полезно. Давайте сделаем чат, где все пользователи видят сообщения друг друга.

Нам нужно:

• Хранить список активных подключений
• Принимать сообщения от одного клиента
• Отправлять их всем остальным

import asyncio
import websockets

# Хранилище подключённых клиентов
connected = set()

async def chat_handler(websocket):
    # Добавляем клиента в список
    connected.add(websocket)
    try:
        async for message in websocket:
            # Отправляем сообщение всем, кроме отправителя
            await asyncio.gather(
                *[client.send(message) for client in connected if client != websocket],
                return_exceptions=True
            )
    except websockets.exceptions.ConnectionClosed:
        # Удаляем клиента при отключении
        connected.remove(websocket)
    finally:
        connected.discard(websocket)

async def main():
    async with websockets.serve(chat_handler, "localhost", 8765):
        await asyncio.Future()

asyncio.run(main())

Теперь, когда один пользователь пишет «Привет!», все остальные видят это. Никаких серверных перезагрузок. Никаких таймаутов. Даже если 50 человек одновременно пишут - сервер справляется, потому что всё асинхронно.

Обратите внимание на asyncio.gather: он отправляет сообщения всем клиентам параллельно. Если один клиент отключился - ошибка не остановит всю систему, потому что мы используем return_exceptions=True. Это важно для стабильности.

Как структурировать сообщения: JSON вместо текста

Отправлять просто текст - это как писать бумажки и бросать их в общий ящик. Никто не знает, кто отправил, когда, и что именно. Лучше использовать JSON. Пример:

{
  "type": "message",
  "user": "Анна",
  "text": "Привет, все!",
  "timestamp": "2026-03-10T12:30:00Z"
}

Тогда клиент может:

• Отображать имя отправителя
• Сортировать сообщения по времени
• Обрабатывать системные события: «пользователь вошёл», «пользователь вышел»

На сервере это выглядит так:

import json

async def chat_handler(websocket):
    connected.add(websocket)
    try:
        async for message in websocket:
            data = json.loads(message)
            if data["type"] == "message":
                # Отправляем всем, кроме отправителя
                await asyncio.gather(
                    *[client.send(message) for client in connected if client != websocket],
                    return_exceptions=True
                )
            elif data["type"] == "join":
                # Уведомляем всех, что кто-то зашёл
                notification = json.dumps({
                    "type": "system",
                    "text": f"{data['user']} присоединился"
                })
                await asyncio.gather(
                    *[client.send(notification) for client in connected],
                    return_exceptions=True
                )
    except websockets.exceptions.ConnectionClosed:
        connected.remove(websocket)
    finally:
        connected.discard(websocket)

Теперь вы можете добавить логику: кто онлайн, кто написал, когда отключился. Это уже полноценный чат.

Масштабирование: Redis и несколько серверов

А что, если один сервер не справляется? Допустим, у вас 10 000 пользователей. Один сервер - это максимум. Нужно несколько серверов. Но как им обмениваться сообщениями?

Решение - Redis. Это быстрая база данных, которая умеет работать как система публикации-подписки (pub/sub). Каждый сервер подписывается на канал chat:messages. Когда один сервер получает сообщение от клиента - он публикует его в Redis. Все остальные серверы получают его и рассылают своим клиентам.

Пример на Python с aioredis:

import aioredis

redis = aioredis.from_url("redis://localhost")

async def pubsub_handler():
    pubsub = redis.pubsub()
    await pubsub.subscribe("chat:messages")
    async for message in pubsub.listen():
        if message["type"] == "message":
            # Отправляем всем клиентам на этом сервере
            await asyncio.gather(
                *[client.send(message["data"]) for client in connected],
                return_exceptions=True
            )

# Запускаем обработчик в фоне
asyncio.create_task(pubsub_handler())

Теперь вы можете запустить 5 серверов на разных машинах. Они все слушают Redis. Любой клиент, подключённый к любому серверу, видит сообщения от всех. Это стандартный подход для высоконагруженных приложений - например, для Telegram или Discord.

Что ещё можно сделать с веб-сокетами

Чат - это только начало. Вот что реально работает на веб-сокетах в 2026 году:

• Игры: многопользовательские онлайн-игры, где движения игроков синхронизируются в реальном времени.
• Трейдинг: обновление цен на акции, криптовалюты, фьючерсы - каждые 50 миллисекунд.
• Мониторинг: логи серверов, статистика использования, алерты - приходят сразу, как только что-то пошло не так.
• Уведомления: новые сообщения, лайки, комментарии - всё без обновления страницы.
• Коллаборация: редактирование документа несколькими людьми одновременно - как Google Docs, только на Python.

Всё это требует одного: постоянного, быстрого, надёжного соединения. И веб-сокеты - лучший инструмент для этого в Python.

Что использовать: websockets, FastAPI, или что-то ещё?

Библиотека websockets - это минималистичный инструмент. Она даёт вам полный контроль. Но если вы пишете веб-приложение, лучше использовать FastAPI. Он встроен в поддержку WebSocket и позволяет легко комбинировать REST API и веб-сокеты в одном приложении.

from fastapi import FastAPI
from fastapi.websockets import WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Echo: {data}")

FastAPI автоматически обрабатывает асинхронность, поддерживает OpenAPI-документацию, работает с Pydantic и легко масштабируется. Если вы делаете что-то серьёзное - начинайте с FastAPI. Если вы учитесь или пишете эксперимент - используйте websockets напрямую.

Ошибки, которые ломают всё

Несколько критических ошибок, которые я видел в реальных проектах:

• Не обрабатывают отключение. Если клиент отключился, а вы не удалили его из списка - сервер будет пытаться отправить ему сообщения в пустоту. Это приводит к утечке памяти.
• Не ограничивают размер сообщений. Злоумышленник может отправить 100 МБ за раз - и сервер упадёт. Добавьте проверку: if len(data) > 1024: raise Exception("Too large").
• Не логируют ошибки. Если что-то сломалось - вы не узнаете. Всегда пишите в лог: logger.error(f"Connection failed: {e}").
• Забывают про CORS и безопасность. WebSocket не защищён по умолчанию. Всегда проверяйте Origin-заголовок, если приложение публичное.

Если вы сделаете всё правильно - ваш сервер будет работать неделями без перезагрузки. Я видел проекты, где WebSocket-сервер на Python работал 8 месяцев без сбоев - и обслуживал 15 000 одновременных пользователей.

Где взять примеры и тестировать

Попробуйте сами:

• Установите: pip install websockets asyncio
• Скопируйте код echo-сервера выше
• Запустите его
• Откройте этот тестовый клиент (или любой другой)
• Пишите - и смотрите, как ответы возвращаются мгновенно

Потом попробуйте сделать чат. Добавьте форму, JavaScript-клиент, и подключите его к вашему серверу. Это займёт 2 часа - и вы получите опыт, который стоит больше, чем десятки курсов.

Асинхронные веб-сокеты - это не фича. Это стандарт. И в Python он работает настолько просто, что вы удивитесь, почему раньше этого не делали.