WebSocket на C#: реальные проекты и продвинутые техники

WebSocket на C#: от прототипа к продакшену — продвинутые техники real-time приложений

Стандартная реализация WebSocket на C# с помощью библиотек вроде System.Net.WebSockets или SignalR решает базовые задачи чатов и уведомлений. Однако, когда требования выходят за рамки "Hello, World", разработчик сталкивается с необходимостью масштабирования, обеспечения безопасности, кастомной маршрутизации и оптимизации производительности. В этой статье мы разберем продвинутые техники работы с WebSocket в .NET, которые позволят вам строить надежные и высоконагруженные real-time системы.

1. Архитектура масштабируемого WebSocket-сервера: паттерны и кластеризация

Одно из главных ограничений стандартного WebSocket — это привязка соединения к конкретному экземпляру процесса. При масштабировании (горизонтальном) сообщение, отправленное на сервер A, не достигнет клиента, подключенного к серверу B. Для решения этой проблемы используются два подхода:

• Централизованный бэкплейн (Backplane): Использование Redis Pub/Sub, RabbitMQ или Azure SignalR Service для обмена сообщениями между серверами.
• Sticky Sessions (липкие сессии): Балансировщик направляет запросы клиента всегда на один и тот же сервер. Менее надежный метод, так как при падении узла все его клиенты теряют соединение.

Рассмотрим пример реализации кастомного бэкплейна на Redis с использованием StackExchange.Redis и System.Net.WebSockets.

using StackExchange.Redis;
using System.Net.WebSockets;
using System.Text;

public class RedisBackplaneManager
{
    private readonly ConnectionMultiplexer _redis;
    private readonly ISubscriber _subscriber;
    private readonly ConcurrentDictionary<string, WebSocket> _connections;

    public RedisBackplaneManager(string redisConnectionString)
    {
        _redis = ConnectionMultiplexer.Connect(redisConnectionString);
        _subscriber = _redis.GetSubscriber();
        _connections = new ConcurrentDictionary<string, WebSocket>();
        
        // Подписываемся на канал "websocket:messages"
        _subscriber.Subscribe("websocket:messages", (channel, message) => {
            var data = Encoding.UTF8.GetString(message);
            // Парсим сообщение (предполагаем формат: "connectionId:payload")
            var parts = data.Split(':', 2);
            if (parts.Length == 2 && _connections.TryGetValue(parts[0], out var ws))
            {
                ws.SendAsync(new ArraySegment<byte>(Encoding.UTF8.GetBytes(parts[1])), 
                    WebSocketMessageType.Text, true, CancellationToken.None);
            }
        });
    }

    public async Task SendToClientAsync(string connectionId, string message)
    {
        // Публикуем в Redis, чтобы сообщение получили все серверы
        var payload = $"{connectionId}:{message}";
        await _subscriber.PublishAsync("websocket:messages", payload);
    }

    public void AddConnection(string id, WebSocket ws) => _connections.TryAdd(id, ws);
    public void RemoveConnection(string id) => _connections.TryRemove(id, out _);
}

2. Кастомный протокол и маршрутизация: WebSocket как RPC-транспорт

Часто необходимо не просто отправлять текст, а вызывать удаленные процедуры (RPC) через WebSocket. Вместо того чтобы вручную парсить JSON и проверять команды, можно построить минималистичный маршрутизатор сообщений. Этот подход используется в SignalR под капотом, но вы можете реализовать его для полного контроля.

public class WebSocketRouter
{
    private readonly Dictionary<string, Func<string, Task>> _routes = new();

    public void On(string action, Func<string, Task> handler)
    {
        _routes[action] = handler;
    }

    public async Task RouteAsync(string rawMessage)
    {
        // Ожидаем формат: {"action":"methodName", "data":{...}}
        var json = JsonDocument.Parse(rawMessage);
        var action = json.RootElement.GetProperty("action").GetString();
        var data = json.RootElement.GetProperty("data").GetRawText();

        if (_routes.TryGetValue(action, out var handler))
        {
            await handler(data);
        }
        else
        {
            // Отправляем ошибку обратно клиенту
            throw new InvalidOperationException($"Unknown action: {action}");
        }
    }
}

// Использование в обработчике WebSocket:
var router = new WebSocketRouter();
router.On("sendMessage", async (data) => {
    var msg = JsonSerializer.Deserialize<MessageDto>(data);
    // Логика отправки сообщения
});
router.On("joinRoom", async (data) => { /* ... */ });

// В цикле обработки сообщений:
var result = await webSocket.ReceiveAsync(buffer, ct);
var incomingMessage = Encoding.UTF8.GetString(buffer.Array, 0, result.Count);
await router.RouteAsync(incomingMessage);

3. Управление состоянием и пульс соединения (Heartbeat)

В продакшен-среде WebSocket-соединения могут разрываться по разным причинам: таймауты прокси, отключение сети, падение клиента. Для своевременного обнаружения "мертвых" соединений необходимо реализовать механизм Heartbeat (keep-alive).

Продвинутая реализация включает:

• Ping/Pong фреймы: Используйте нативные WebSocket Ping/Pong, если это поддерживается библиотекой (например, WebSocket.SendAsync с флагом WebSocketMessageType.Ping).
• Таймеры на стороне сервера: Если клиент не прислал Pong в ответ на Ping в течение N секунд, соединение закрывается.
• Graceful shutdown: При закрытии приложения необходимо дождаться завершения всех обработчиков и корректно закрыть соединения.

public class HeartbeatMiddleware
{
    private readonly TimeSpan _heartbeatInterval = TimeSpan.FromSeconds(30);
    private readonly TimeSpan _timeout = TimeSpan.FromSeconds(10);
    private readonly ILogger _logger;

    public async Task StartHeartbeatAsync(WebSocket socket, CancellationToken ct)
    {
        using var timer = new PeriodicTimer(_heartbeatInterval);
        while (await timer.WaitForNextTickAsync(ct))
        {
            if (socket.State != WebSocketState.Open) break;

            try
            {
                var pingBuffer = new ArraySegment<byte>(Encoding.UTF8.GetBytes("ping"));
                await socket.SendAsync(pingBuffer, WebSocketMessageType.Text, true, ct);

                // Ждем ответный pong в отдельном потоке (упрощенно)
                var pongTask = socket.ReceiveAsync(new ArraySegment<byte>(new byte[1024]), ct);
                if (await Task.WhenAny(pongTask, Task.Delay(_timeout, ct)) != pongTask)
                {
                    _logger.LogWarning("Heartbeat timeout. Closing connection.");
                    await socket.CloseAsync(WebSocketCloseStatus.NormalClosure, "Timeout", ct);
                    break;
                }
            }
            catch (