ASP.NET Core WebSocket集群實現思路詳解

Posted on 2022-11-09 by WalkonNet

前言

提到WebSocket相信大傢都聽說過，它的初衷是為瞭解決客戶端瀏覽器與服務端進行雙向通信，是在單個TCP連接上進行全雙工通訊的協議。在沒有WebSocket之前隻能通過瀏覽器到服務端的請求應答模式比如輪詢，來實現服務端的變更響應到客戶端，現在服務端也可以主動發送數據到客戶端瀏覽器。WebSocket協議和Http協議平行，都屬於TCP/IP四層模型中的第四層應用層。由於WebSocket握手階段采用HTTP協議,所以也需要進行跨域處理。它的協議標識是ws或wss對應瞭常規標識和安全通信協議標識。本文重點並不是介紹WebSocket協議相關，而是提供一種基於ASP.NET Core原生WebSocket的方式實現集群的實現思路。關於這套思路其實很早之前我就構思過瞭，隻是之前一直沒有系統的整理出來，本篇文章就來和大傢分享一下，由於主要是提供一種思路，所以涉及到具體細節或者業務相關的可能沒有體現出來，還望大傢理解。

實現

咱們的重點關鍵字就是兩個WebSocket和集群，實現的框架便是基於ASP.NET Core,我也基於golang實現瞭一套,本文涉及到的相關源碼和golang版本的實現都已上傳至我的github,具體倉庫地址可以轉到文末自行跳轉到#示例源碼中查看。既然涉及到集群，這裡咱們就用nginx作為反向代理，來搭建一個集群實例。大致的示例結構如下圖所示

redis在這裡扮演的角色呢，是用來處理Server端的消息相互傳遞用的，主要是使用的redis的pub/sub功能來實現的，這裡便涉及到幾個核心問題

首先，集群狀態每個用戶被分發到具體的哪臺服務器上是不得而知的
其次，處在不同Server端的不同用戶間的相互通信是需要一個傳遞媒介
最後，針對不同的場景比如單發消息、分組消息、全部通知等要有不同的處理策略

這裡需要考慮的是，如果需要搭建實時通信服務器的話，需要註意集群的隔離性，主要是和核心業務進行隔離，畢竟WebSocket需要保持長鏈接、且消息的大小需要評估。

上面提到瞭redis的主要功能就是用來傳遞消息用的，畢竟每個server服務器是無狀態的。這當然不是必須的，任何可以進行消息分發的中間件都可以，比如消息隊列rabbitmq、kafka、rocketmq、mqtt等，甚至隻要能把要處理的消息存儲起來都可以比如緩存甚至是關系型數據庫等等。這壓力使用redis主要是因為操作起來簡單、輕量級、靈活，讓大傢關註點在思路上，而不是使用中案件的代碼上。

nginx配置

通過上面的圖我們可以看到，我們這裡構建集群示例使用的nginx，如果讓nginx支持WebSocket的話，需要額外的配置，這個在網上有很多相關的文章介紹，這裡就來列一下咱們示例的nginx配置，在配置文件nginx.conf裡

//上遊服務器地址也就是websocket服務的真實地址
upstream wsbackend {
    server 127.0.0.1:5001;
    server 127.0.0.1:5678;
}
server {
    listen       5000;
    server_name  localhost;
    location ~/chat/{
        //upstream地址
        proxy_pass http://wsbackend;
        proxy_connect_timeout 60s; 
        proxy_read_timeout 3600s;
        proxy_send_timeout 3600s;
        //記得轉發避免踩坑
        proxy_set_header Host $host;
        proxy_http_version 1.1; 
        //http升級成websocket協議的頭標識
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "Upgrade";
    }
}

這套配置呢，在搜索引擎上能收到很多，不過不妨礙我把使用的粘貼出來。這一套親測有效，也是我使用的配置，請放心使用。個人認為如果是線上環境采用的負載均衡策略可以選擇ip_hash的方式，保證同一個ip的客戶端用戶可以分發到一臺WebSocket實例中去，這樣的話能盡量避免使用redis的用戶頻道做消息傳遞。好瞭，接下來準備開始展示具體實現的代碼瞭。

一對一發送

首先介紹的就是一對一發送的情況，也就是我把消息發給你，聊天的時候私聊的情況。這裡呢涉及到兩種情況

如果你需要通信的客戶端和你連接在一個Server端裡，這樣的話可以直接在鏈接裡找到這個端的通信實例直接發送。
如果你需要通信的客戶端和你不在一個Server端裡，這個時候咱們就需要借助redis的pub/sub的功能，把消息傳遞給另一個Server端。

咱們通過一張圖大致的展示一下它的工作方式

解釋一下，每個客戶端註冊到WebSocket服務裡的時候會在redis裡訂閱一個user:用戶唯一標識的頻道，這個頻道用於接收和當前WebSocket連接不在一個服務端的其他WebSocket發送過來的消息。

每次發送消息的時候你會知道你要發送給誰，不在當前服務器的話則發送到redis的user:用戶唯一標識頻道，這樣的話目標WebSocket就能收到消息瞭。

首先是註入相關的依賴項，這裡我使用的redis客戶端是freeredis，主要是因為操作起來簡單,具體實現代碼如下

var builder = WebApplication.CreateBuilder(args);
//註冊freeredis
builder.Services.AddSingleton(provider => {
    var logger = provider.GetService<ILogger<WebSocketChannelHandler>>();
    RedisClient cli = new RedisClient("127.0.0.1:6379");
    cli.Notice += (s, e) => logger?.LogInformation(e.Log);
    return cli;
});
//註冊WebSocket具體操作的類
builder.Services.AddSingleton<WebSocketHandler>();
builder.Services.AddControllers();
var app = builder.Build();
var webSocketOptions = new WebSocketOptions
{
    KeepAliveInterval = TimeSpan.FromMinutes(2)
};
//註冊WebSocket中間件
app.UseWebSockets(webSocketOptions);
app.MapGet("/", () => "Hello World!");
app.MapControllers();
app.Run();

接下來我們定義一個Controller用來處理WebSocket請求

public class WebSocketController : ControllerBase
{
    private readonly ILogger<WebSocketController> _logger;
    private readonly WebSocketHandler _socketHandler;
    public WebSocketController(ILogger<WebSocketController> logger, WebSocketHandler socketHandler, WebSocketChannelHandler webSocketChannelHandler)
    {
        _logger = logger;
        _socketHandler = socketHandler;
    }
    //這裡的id代表當前連接的客戶端唯一標識比如用戶唯一標識
    [HttpGet("/chat/user/{id}")]
    public async Task ChatUser(string id)
    {
        //判斷是否是WebSocket請求
        if (HttpContext.WebSockets.IsWebSocketRequest)
        {
            _logger.LogInformation($"user:{id}-{Request.HttpContext.Connection.RemoteIpAddress}:{Request.HttpContext.Connection.RemotePort} join");
            var webSocket = await HttpContext.WebSockets.AcceptWebSocketAsync();
            //處理請求相關
            await _socketHandler.Handle(id, webSocket);
        }
        else
        {
            HttpContext.Response.StatusCode = StatusCodes.Status400BadRequest;
        }
    }
}

這裡的WebSocketHandler是用來處理具體邏輯用的，咱們看一下相關代碼

public class WebSocketHandler:IDisposable
{
    //存儲當前服務用戶的集合
    private readonly UserConnection UserConnection = new();
    //redis頻道前綴
    private readonly string userPrefix = "user:";
    //用戶對應的redis頻道
    private readonly ConcurrentDictionary<string, IDisposable> _disposables = new();
    private readonly ILogger<WebSocketHandler> _logger;
    //redis客戶端
    private readonly RedisClient _redisClient;
    public WebSocketHandler(ILogger<WebSocketHandler> logger, RedisClient redisClient)
    {
        _logger = logger;
        _redisClient = redisClient;
    }
    public async Task Handle(string id, WebSocket webSocket)
    {
        //把當前用戶連接存儲起來
        _ = UserConnection.GetOrAdd(id, webSocket);
        //訂閱一個當前用戶的頻道
        await SubMsg($"{userPrefix}{id}");
        var buffer = new byte[1024 * 4];
        //接收發送過來的消息，這個方法是阻塞的，如果沒收到消息則一直阻塞
        var receiveResult = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
        //循環接收消息
        while (webSocket.State == WebSocketState.Open)
        {
            try
            {
                //因為緩沖區長度是固定的所以要獲取實際長度
                string msg = Encoding.UTF8.GetString(buffer[..receiveResult.Count]).TrimEnd('\0');
                //接收的到消息轉換成實體
                MsgBody msgBody = JsonConvert.DeserializeObject<MsgBody>(msg);
                //發送到其他客戶端的數據
                byte[] sendByte = Encoding.UTF8.GetBytes($"user {id} send:{msgBody.Msg}");
                _logger.LogInformation($"user {id} send:{msgBody.Msg}");
                //判斷目標客戶端是否在當前當前服務，如果在當前服務直接紮到目標連接直接發送
                if (UserConnection.TryGetValue(msgBody.Id, out var targetSocket))
                {
                    if (targetSocket.State == WebSocketState.Open)
                    {
                        await targetSocket.SendAsync(new ArraySegment<byte>(sendByte, 0, sendByte.Length), receiveResult.MessageType, true, CancellationToken.None);
                    }
                }
                else
                {
                    //如果要發送的目標端不在當前服務，則發送給目標redis端的頻道
                    ChannelMsgBody channelMsgBody = new ChannelMsgBody { FromId = id, ToId = msgBody.Id, Msg = msgBody.Msg };
                    //目標的redis頻道
                    _redisClient.Publish($"{userPrefix}{msgBody.Id}", JsonConvert.SerializeObject(channelMsgBody));
                }
                //繼續阻塞循環接收消息
                receiveResult = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, ex.Message);
                break;
            }
        }
        //循環結束意味著當前端已經退出
        //從當前用戶的集合移除當前用戶
        _ = UserConnection.TryRemove(id, out _);
        //關閉當前WebSocket連接
        await webSocket.CloseAsync(receiveResult.CloseStatus.Value, receiveResult.CloseStatusDescription, CancellationToken.None);
        //在當前訂閱集合移除當前用戶
        _disposables.TryRemove($"{userPrefix}{id}", out var disposable);
        //關閉當前用戶的通道
        disposable.Dispose();
    }
    private async Task SubMsg(string channel)
    {
        //訂閱當前用戶頻道
        var sub = _redisClient.Subscribe(channel,  async (channel, data) => {
            //接收過來當前頻道數據，說明發送端不在當前服務
            ChannelMsgBody msgBody = JsonConvert.DeserializeObject<ChannelMsgBody>(data.ToString());
            byte[] sendByte = Encoding.UTF8.GetBytes($"user {msgBody.FromId} send:{msgBody.Msg}");
            //在當前服務找到目標的WebSocket連接並發送消息
            if (UserConnection.TryGetValue(msgBody.ToId, out var targetSocket))
            {
                if (targetSocket.State == WebSocketState.Open)
                {
                    await targetSocket.SendAsync(new ArraySegment<byte>(sendByte, 0, sendByte.Length), WebSocketMessageType.Text, true, CancellationToken.None);
                }
            }
        });
        //把redis訂閱頻道添加到集合中
        _disposables.TryAdd(channel, sub);
    }
    //程序退出的時候取消當前服務訂閱的redis頻道
    public void Dispose()
    {
        foreach (var disposable in _disposables)
        {
            disposable.Value.Dispose();
        }
        _disposables.Clear();
    }
}

這裡涉及到幾個輔助相關的類，其中UserConnection類是存儲註冊到當前服務的連接，MsgBody類用來接受客戶端發送過來的消息，ChannelMsgBody是用來發送redis頻道的相關消息，因為要把相關消息通過redis發佈出去，咱們列一下這幾個類的相關代碼

//註冊到當前服務的連接
public class UserConnection : IEnumerable<KeyValuePair<string, WebSocket>>
{
    //存儲用戶唯一標識和WebSocket的對應關系
    private ConcurrentDictionary<string, WebSocket> _users = new ConcurrentDictionary<string, WebSocket>();
    //當前服務的用戶數量
    public int Count => _users.Count;
    public WebSocket GetOrAdd(string userId, WebSocket webSocket)
    {
        return _users.GetOrAdd(userId, webSocket);
    }
    public bool TryGetValue(string userId, out WebSocket webSocket)
    {
        return _users.TryGetValue(userId, out webSocket);
    }
    public bool TryRemove(string userId, out WebSocket webSocket)
    {
        return _users.TryRemove(userId, out webSocket);
    }
    public void Clear()
    {
        _users.Clear();
    }
    public IEnumerator<KeyValuePair<string, WebSocket>> GetEnumerator()
    {
        return _users.GetEnumerator();
    }
    IEnumerator IEnumerable.GetEnumerator()
    {
        return this.GetEnumerator();
    }
}
//客戶端消息
public class MsgBody
{
    //目標用戶標識
    public string Id { get; set; }
    //要發送的消息
    public string Msg { get; set; }
}
//頻道訂閱消息
public class ChannelMsgBody
{
    //用戶標識
    public string FromId { get; set; }
    //目標用戶標識，也就是要發送給誰
    public string ToId { get; set; }
    //要發送的消息
    public string Msg { get; set; }
}

這樣的話關於一對一發送消息的相關邏輯就實現完成瞭，啟動兩個Server端，由於nginx默認的負載均衡策略是輪詢，所以註冊兩個用戶的話會被分發到不同的服務裡去

用Postman連接三個連接唯一標識分別是1、2、3，模擬一下消息發送，效果如下,發送效果

接收效果

群組發送

上面我們展示瞭一對一發送的情況，接下來我們來看一下，群組發送的情況。群組發送的話就是隻要大傢都加入一個群組，隻要客戶端在群組裡發送一條消息，則註冊到當前群組內的所有客戶端都可以收到消息。相對於一對一的情況就是如果當前WebSocket服務端如果存在用戶加入某個群組，則當前當前WebSocket服務端則可以訂閱一個group:群組唯一標識的redis頻道，集群中的其他WebSocket服務器通過這個redis頻道接收群組消息，通過一張圖描述一下

群組的實現方式相對於一對一要簡單一點

發送端可以不用考慮當前服務中的客戶端連接，一股腦的交給redis把消息發佈出去
如果有WebSocket服務中的用戶訂閱瞭當前分組則可以接受消息，獲取組內的用戶循環發送消息

展示一下代碼實現的方式,首先是定義一個action用於表示群組的相關場景

//包含兩個標識一個是組別標識一個是註冊到組別的用戶
[HttpGet("/chat/group/{groupId}/{userId}")]
public async Task ChatGroup(string groupId, string userId)
{
    if (HttpContext.WebSockets.IsWebSocketRequest)
    {
        _logger.LogInformation($"group:{groupId} user:{userId}-{Request.HttpContext.Connection.RemoteIpAddress}:{Request.HttpContext.Connection.RemotePort} join");
        var webSocket = await HttpContext.WebSockets.AcceptWebSocketAsync();
        //調用HandleGroup處理群組相關的消息
        await _socketHandler.HandleGroup(groupId, userId, webSocket);
    }
    else
    {
        HttpContext.Response.StatusCode = StatusCodes.Status400BadRequest;
    }
}

接下來看一下HandleGroup的相關邏輯，還是在WebSocketHandler類中，看一下代碼實現

public class WebSocketHandler:IDisposable
{
    private readonly UserConnection UserConnection = new();
    private readonly GroupUser GroupUser = new();
    private readonly SemaphoreSlim _lock = new(1, 1);
    private readonly ConcurrentDictionary<string, IDisposable> _disposables = new();
    private readonly string groupPrefix = "group:";
    private readonly ILogger<WebSocketHandler> _logger;
    private readonly RedisClient _redisClient;
    public WebSocketHandler(ILogger<WebSocketHandler> logger, RedisClient redisClient)
    {
        _logger = logger;
        _redisClient = redisClient;
    }
    public async Task HandleGroup(string groupId, string userId, WebSocket webSocket)
    {
        //因為群組的集合可能會存在很多用戶一起訪問所以限制訪問數量
        await _lock.WaitAsync();
        //初始化群組容器 群唯一標識為key 群員容器為value
        var currentGroup = GroupUser.Groups.GetOrAdd(groupId, new UserConnection { });
        //當前用戶加入當前群組
        _ = currentGroup.GetOrAdd(userId, webSocket);
        //隻有有當前WebSocket服務的第一個加入當前組的時候才去訂閱群組頻道
        //如果不限制的話則會出現如果當前WebSocket服務有多個用戶在一個組內則會重復收到redis消息
        if (currentGroup.Count == 1)
        {
            //訂閱redis頻道
            await SubGroupMsg($"{groupPrefix}{groupId}");
        }
        _lock.Release();
        var buffer = new byte[1024 * 4];
        //阻塞接收WebSocket消息
        var receiveResult = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
        //服務不退出的話則一直等待接收
        while (webSocket.State == WebSocketState.Open)
        {
            try
            {
                string msg = Encoding.UTF8.GetString(buffer[..receiveResult.Count]).TrimEnd('\0');
                _logger.LogInformation($"group 【{groupId}】 user 【{userId}】 send:{msg}");
                //組裝redis頻道發佈的消息，目標為群組標識
                ChannelMsgBody channelMsgBody = new ChannelMsgBody { FromId = userId, ToId = groupId, Msg = msg };
                //通過redis發佈消息
                _redisClient.Publish($"{groupPrefix}{groupId}", JsonConvert.SerializeObject(channelMsgBody));
                receiveResult = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, ex.Message);
                break;
            }
        }
        //如果客戶端退出則在當前群組集合刪除當前用戶
        _ = currentGroup.TryRemove(userId, out _);
        await webSocket.CloseAsync(receiveResult.CloseStatus.Value, receiveResult.CloseStatusDescription, CancellationToken.None);
    }
    private async Task SubGroupMsg(string channel)
    {
        var sub = _redisClient.Subscribe(channel, async (channel, data) => {
            ChannelMsgBody msgBody = JsonConvert.DeserializeObject<ChannelMsgBody>(data.ToString());
            byte[] sendByte = Encoding.UTF8.GetBytes($"group 【{msgBody.ToId}】 user 【{msgBody.FromId}】 send:{msgBody.Msg}");
            //在當前WebSocket服務器找到當前群組裡的用戶
            GroupUser.Groups.TryGetValue(msgBody.ToId, out var currentGroup);
            //循環當前WebSocket服務器裡的用戶發送消息
            foreach (var user in currentGroup)
            {
                //不用給自己發送瞭
                if (user.Key == msgBody.FromId)
                {
                    continue;
                }
                if (user.Value.State == WebSocketState.Open)
                {
                    await user.Value.SendAsync(new ArraySegment<byte>(sendByte, 0, sendByte.Length), WebSocketMessageType.Text, true, CancellationToken.None);
                }
            }
        });
        //把當前頻道加入訂閱集合
        _disposables.TryAdd(channel, sub);
    }
}

這裡涉及到瞭GroupUser類，是來存儲群組和群組用戶的對應關系的，定義如下

public class GroupUser
{
    //key為群組的唯一標識
    public ConcurrentDictionary<string, UserConnection> Groups = new ConcurrentDictionary<string, UserConnection>();
}

演示一下把兩個用戶添加到一個群組內，然後發送接收消息的場景，用戶u1發送

用戶u2接收

發送所有人

發送給所有用戶的邏輯比較簡單，不用考慮到用戶限制，隻要用戶連接到瞭WebSocket集群則都可以接收到這個消息，大致工作方式如下圖所示

這個比較簡單，咱們直接看實現代碼，首先是定義一個地址，用於發佈消息

//把用戶註冊進去
[HttpGet("/chat/all/{id}")]
public async Task ChatAll(string id)
{
    if (HttpContext.WebSockets.IsWebSocketRequest)
    {
        _logger.LogInformation($"all user:{id}-{Request.HttpContext.Connection.RemoteIpAddress}:{Request.HttpContext.Connection.RemotePort} join");
        var webSocket = await HttpContext.WebSockets.AcceptWebSocketAsync();
        await _socketHandler.HandleAll(id, webSocket);
    }
    else
    {
        HttpContext.Response.StatusCode = StatusCodes.Status400BadRequest;
    }
}

具體的實現邏輯還是在HandleGroup類裡，是HandleAll方法，看一下具體實現

public class WebSocketHandler:IDisposable
{
    private readonly UserConnection AllConnection = new();
    private readonly ConcurrentDictionary<string, IDisposable> _disposables = new();
    private readonly string all = "all";
    private readonly ILogger<WebSocketHandler> _logger;
    private readonly RedisClient _redisClient;
    public WebSocketHandler(ILogger<WebSocketHandler> logger, RedisClient redisClient)
    {
        _logger = logger;
        _redisClient = redisClient;
    }
    public async Task HandleAll(string id, WebSocket webSocket)
    {
        await _lock.WaitAsync();
        //把用戶加入用戶集合
        _ = AllConnection.GetOrAdd(id, webSocket);
        //WebSocket集群中的每個服務隻定義一次
        if (AllConnection.Count == 1)
        {
            await SubAllMsg(all);
        }
        _lock.Release();
        var buffer = new byte[1024 * 4];
        //阻塞接收信息
        var receiveResult = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
        while (webSocket.State == WebSocketState.Open)
        {
            try
            {
                string msg = Encoding.UTF8.GetString(buffer[..receiveResult.Count]).TrimEnd('\0');
                _logger.LogInformation($"user {id} send:{msg}");
                //獲取接收信息
                ChannelMsgBody channelMsgBody = new ChannelMsgBody { FromId = id, Msg = msg };
                //把消息通過redis發佈到集群中的其他服務
                _redisClient.Publish(all, JsonConvert.SerializeObject(channelMsgBody));
                receiveResult = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, ex.Message);
                break;
            }
        }
        //用戶退出則刪除集合中的當前用戶信息
        _ = AllConnection.TryRemove(id, out _);
        await webSocket.CloseAsync(receiveResult.CloseStatus.Value, receiveResult.CloseStatusDescription, CancellationToken.None);
    }
    private async Task SubAllMsg(string channel)
    {
        var sub = _redisClient.Subscribe(channel, async (channel, data) => {
            ChannelMsgBody msgBody = JsonConvert.DeserializeObject<ChannelMsgBody>(data.ToString());
            byte[] sendByte = Encoding.UTF8.GetBytes($"user 【{msgBody.FromId}】 send all:{msgBody.Msg}");
            //接收到消息後遍歷用戶集合把消息發送給所有用戶
            foreach (var user in AllConnection)
            {   
                //如果包含當前用戶跳過
                if (user.Key == msgBody.FromId)
                {
                    continue;
                }
                if (user.Value.State == WebSocketState.Open)
                {
                    await user.Value.SendAsync(new ArraySegment<byte>(sendByte, 0, sendByte.Length), WebSocketMessageType.Text, true, CancellationToken.None);
                }
            }
        });
        _disposables.TryAdd(channel, sub);
    }
}

效果在這裡就不展示瞭，和群組的效果是類似的，隻是一個是部分用戶，一個是全部的用戶。

整合到一起

上面我們分別展示瞭一對一、群組、所有人的場景，但是實際使用的時候，每個用戶隻需要註冊到WebSocket集群一次也就是保持一個連接即可，而不是一對一一個連接、註冊群組一個連接、所有消息的時候一個連接。所以我們需要把上面的演示整合一下，一個用戶隻需要連接到WebSocket集群一次即可，至於發送給誰，加入什麼群組，接收全部消息等都是連接後通過一些標識區分的，而不必每個類型的操作都註冊一次，就和微信和QQ一樣我隻要登錄瞭即可，至於其他操作都是靠數據標識區分的。接下來咱們就整合一下代碼達到這個效果，大致的思路是

用戶連接到WebSocket集群，把用戶和連接保存到當前WebSocket服務器的用戶集合中去。
一對一發送的時候，隻需要在具體的服務器中找到具體的客戶端發送消息
群組的時候，先把當前用戶標識加入群組集合即可，接收消息的時候根據群組集合裡的用戶標識去用戶集合裡去拿具體的WebSocket連接發送消息
全員消息的時候，直接遍歷集群中的每個WebSocket服務裡的用戶集合裡的WebSocket連接訓話發送消息

這樣的話就保證瞭每個客戶端用戶在集群中隻會綁定一個連接，首先還是單獨定義一個action，用於讓客戶端用戶連接上來，具體實現代碼如下所示

public class WebSocketChannelController : ControllerBase
{
    private readonly ILogger<WebSocketController> _logger;
    private readonly WebSocketChannelHandler _webSocketChannelHandler;
    public WebSocketChannelController(ILogger<WebSocketController> logger, WebSocketChannelHandler webSocketChannelHandler)
    {
        _logger = logger;
        _webSocketChannelHandler = webSocketChannelHandler;
    }
    //隻需要把當前用戶連接到服務即可
    [HttpGet("/chat/channel/{id}")]
    public async Task Channel(string id)
    {
        if (HttpContext.WebSockets.IsWebSocketRequest)
        {
            _logger.LogInformation($"user:{id}-{Request.HttpContext.Connection.RemoteIpAddress}:{Request.HttpContext.Connection.RemotePort} join");
            var webSocket = await HttpContext.WebSockets.AcceptWebSocketAsync();
            await _webSocketChannelHandler.HandleChannel(id, webSocket);
        }
        else
        {
            HttpContext.Response.StatusCode = StatusCodes.Status400BadRequest;
        }
    }
}

接下來看一下WebSocketChannelHandler類的HandleChannel方法實現，用於處理不同的消息，比如一對一、群組、全員消息等不同類型的消息

public class WebSocketChannelHandler : IDisposable
{
    //用於存儲當前WebSocket服務器鏈接上來的所有用戶對應關系
    private readonly UserConnection UserConnection = new();
    //用於存儲群組和用戶關系，用戶集合采用HashSet保證每個用戶隻加入一個群組一次
    private readonly ConcurrentDictionary<string, HashSet<string>> GroupUser = new ConcurrentDictionary<string, HashSet<string>>();
    private readonly SemaphoreSlim _lock = new(1, 1);
    //存放redis訂閱實例
    private readonly ConcurrentDictionary<string, IDisposable> _disposables = new();
    //一對一redis頻道前綴
    private readonly string userPrefix = "user:";
    //群組redis頻道前綴
    private readonly string groupPrefix = "group:";
    //全員redis頻道
    private readonly string all = "all";
    private readonly ILogger<WebSocketHandler> _logger;
    private readonly RedisClient _redisClient;
    public WebSocketChannelHandler(ILogger<WebSocketHandler> logger, RedisClient redisClient)
    {
        _logger = logger;
        _redisClient = redisClient;
    }
    public async Task HandleChannel(string id, WebSocket webSocket)
    {
        await _lock.WaitAsync();
        //每次連接進來就添加到用戶集合
        _ = UserConnection.GetOrAdd(id, webSocket);
        //每個WebSocket服務實例隻需要訂閱一次全員消息頻道
        await SubMsg($"{userPrefix}{id}");
        if (UserConnection.Count == 1)
        {
            await SubAllMsg(all);
        }
        _lock.Release();
        var buffer = new byte[1024 * 4];
        //接收客戶端消息
        var receiveResult = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
        while (webSocket.State == WebSocketState.Open)
        {
            try
            {
                string msg = Encoding.UTF8.GetString(buffer[..receiveResult.Count]).TrimEnd('\0');
                //讀取客戶端消息
                ChannelData channelData = JsonConvert.DeserializeObject<ChannelData>(msg);
                //判斷消息類型
                switch (channelData.Method)
                {
                    //一對一
                    case "One":
                        await HandleOne(id, channelData.MsgBody, receiveResult);
                        break;
                    //把用戶加入群組
                    case "UserGroup":
                        await AddUserGroup(id, channelData.Group, webSocket);
                        break;
                    //處理群組消息
                    case "Group":
                        await HandleGroup(channelData.Group, id, webSocket, channelData.MsgBody);
                        break;
                    //處理全員消息
                    default:
                        await HandleAll(id, channelData.MsgBody);
                        break;
                }
                receiveResult = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, ex.Message);
                break;
            }
        }
        await webSocket.CloseAsync(receiveResult.CloseStatus.Value, receiveResult.CloseStatusDescription, CancellationToken.None);
        //在群組中移除當前用戶
        foreach (var users in GroupUser.Values)
        {
            lock (users)
            {
                users.Remove(id);
            }
        }
        //當前客戶端用戶退出則移除連接
        _ = UserConnection.TryRemove(id, out _);
        //取消用戶頻道訂閱
        _disposables.Remove($"{userPrefix}{id}", out var sub);
        sub?.Dispose();
    }
    public void Dispose()
    {
        foreach (var disposable in _disposables)
        {
            disposable.Value.Dispose();
        }
        _disposables.Clear();
    }
}

這裡涉及到瞭ChannelData類是用於接收客戶端消息的類模板，具體定義如下

public class ChannelData
{
    //消息類型 比如一對一 群組 全員
    public string Method { get; set; }
    //群組標識
    public string Group { get; set; }
    //消息體
    public object MsgBody { get; set; }
}

類中並不會包含當前用戶信息，因為連接到當前服務的時候已經提供瞭客戶端唯一標識。結合上面的處理代碼我們可以看出，客戶端用戶連接到WebSocket實例之後，先註冊當前用戶的redis訂閱頻道並且當前實例僅註冊一次全員消息的redis頻道，用於處理非當前實例註冊客戶端的一對一消息處理和全員消息處理，然後等待接收客戶端消息，根據客戶端消息的消息類型來判斷是進行一對一、群組、或者全員的消息類型處理，它的工作流程入下圖所示

由代碼和上面的流程圖可知，它根據不同的標識去處理不同類型的消息，接下來我們可以看下每種消息類型的處理方式。

一對一處理

首先是一對一的消息處理情況，看一下具體的處理邏輯，首先是一對一發佈消息

 private async Task HandleOne(string id, object msg, WebSocketReceiveResult receiveResult)
 {
    MsgBody msgBody = JsonConvert.DeserializeObject<MsgBody>(JsonConvert.SerializeObject(msg));
    byte[] sendByte = Encoding.UTF8.GetBytes($"user {id} send:{msgBody.Msg}");
    _logger.LogInformation($"user {id} send:{msgBody.Msg}");
    //判斷目標用戶是否在當前WebSocket服務器
    if (UserConnection.TryGetValue(msgBody.Id, out var targetSocket))
    {
        if (targetSocket.State == WebSocketState.Open)
        {
            await targetSocket.SendAsync(new ArraySegment<byte>(sendByte, 0, sendByte.Length), receiveResult.MessageType, true, CancellationToken.None);
        }
    }
    else
    {
        //如果不在當前服務器，則直接把消息發佈到具體的用戶頻道去，由具體用戶去訂閱
        ChannelMsgBody channelMsgBody = new ChannelMsgBody { FromId = id, ToId = msgBody.Id, Msg = msgBody.Msg };
        _redisClient.Publish($"{userPrefix}{msgBody.Id}", JsonConvert.SerializeObject(channelMsgBody));
    }
}

接下來是用於處理訂閱其他用戶發送過來消息的邏輯，這個和整合之前的邏輯是一致的，在當前服務器中找到用戶對應的連接，發送消息

private async Task SubMsg(string channel)
{
    var sub = _redisClient.Subscribe(channel, async (channel, data) =>
    {
        ChannelMsgBody msgBody = JsonConvert.DeserializeObject<ChannelMsgBody>(data.ToString());
        byte[] sendByte = Encoding.UTF8.GetBytes($"user {msgBody.FromId} send:{msgBody.Msg}");
        if (UserConnection.TryGetValue(msgBody.ToId, out var targetSocket))
        {
            if (targetSocket.State == WebSocketState.Open)
            {
                await targetSocket.SendAsync(new ArraySegment<byte>(sendByte, 0, sendByte.Length), WebSocketMessageType.Text, true, CancellationToken.None);
            }
            else
            {
                _ = UserConnection.TryRemove(msgBody.FromId, out _);
            }
        }
    });
    //把訂閱實例加入集合
    _disposables.TryAdd(channel, sub);
}

如果給某個用戶發送消息則可以使用如下的消息格式

{"Method":"One", "MsgBody":{"Id":"2","Msg":"Hello"}}

Method為One代表著是私聊一對一的情況，消息體內Id為要發送給的具體用戶標識和消息體。

群組處理

接下來看群組處理方式，這個和之前的邏輯是有出入的，首先是用戶要先加入到某個群組然後才能接收群組消息或者在群組中發送消息，之前是一個用戶對應多個連接，整合瞭之後集群中每個用戶隻關聯唯一的一個WebSocket連接，首先看用戶加入群組的邏輯

private async Task AddUserGroup(string user, string group, WebSocket webSocket)
{
    //獲取群組信息
    var currentGroup = GroupUser.GetOrAdd(group, new HashSet<string>());
    lock (currentGroup)
    {
       //把用戶標識加入當前組
        _ = currentGroup.Add(user);
    }
    //每個組的redis頻道，在每臺WebSocket服務器實例隻註冊一次訂閱
    if (currentGroup.Count == 1)
    {
        //訂閱當前組消息
        await SubGroupMsg($"{groupPrefix}{group}");
    }
    string addMsg = $"user 【{user}】 add  to group 【{group}】";
    byte[] sendByte = Encoding.UTF8.GetBytes(addMsg);
    await webSocket.SendAsync(new ArraySegment<byte>(sendByte, 0, sendByte.Length), WebSocketMessageType.Text, true, CancellationToken.None);
    //如果有用戶加入群組，則通知其他群成員
    ChannelMsgBody channelMsgBody = new ChannelMsgBody { FromId = user, ToId = group, Msg = addMsg };
    _redisClient.Publish($"{groupPrefix}{group}", JsonConvert.SerializeObject(channelMsgBody));
}

用戶想要在群組內發消息，則必須先加入到一個具體的群組內，具體的加入群組的格式如下

{"Method":"UserGroup", "Group":"g1"}

Method為UserGroup代表著用戶加入群組的業務類型，Group代表著你要加入的群組唯一標識。接下來就看下，用戶發送群組消息的邏輯瞭

private async Task HandleGroup(string groupId, string userId, WebSocket webSocket, object msgBody)
{
    //判斷群組是否存在
    var hasValue = GroupUser.TryGetValue(groupId, out var users);
    if (!hasValue)
    {
        byte[] sendByte = Encoding.UTF8.GetBytes($"group【{groupId}】 not exists");
        await webSocket.SendAsync(new ArraySegment<byte>(sendByte, 0, sendByte.Length), WebSocketMessageType.Text, true, CancellationToken.None);
        return;
    }
    //隻有加入到當前群組，才能在群組內發送消息
    if (!users.Contains(userId))
    {
        byte[] sendByte = Encoding.UTF8.GetBytes($"user 【{userId}】 not in 【{groupId}】");
        await webSocket.SendAsync(new ArraySegment<byte>(sendByte, 0, sendByte.Length), WebSocketMessageType.Text, true, CancellationToken.None);
        return;
    }
    _logger.LogInformation($"group 【{groupId}】 user 【{userId}】 send:{msgBody}");
    //發送群組消息
    ChannelMsgBody channelMsgBody = new ChannelMsgBody { FromId = userId, ToId = groupId, Msg = msgBody.ToString() };
    _redisClient.Publish($"{groupPrefix}{groupId}", JsonConvert.SerializeObject(channelMsgBody));
}

加入群組之後則可以發送和接收群組內的消息瞭，給群組發送消息的格式如下

{"Method":"Group", "Group":"g1", "MsgBody":"Hi All"}

Method為Group代表著用戶加入群組的業務類型，Group則代表你要發送到具體的群組的唯一標識，MsgBody則是發送到群組內的消息。最後再來看下訂閱群組內消息的情況，也就是處理群組消息的邏輯

private async Task SubGroupMsg(string channel)
{
    var sub = _redisClient.Subscribe(channel, async (channel, data) =>
    {
        //接收群組訂閱消息
        ChannelMsgBody msgBody = JsonConvert.DeserializeObject<ChannelMsgBody>(data.ToString());
        byte[] sendByte = Encoding.UTF8.GetBytes($"group 【{msgBody.ToId}】 user 【{msgBody.FromId}】 send:{msgBody.Msg}");
        //獲取當前服務器實例中當前群組的所有用戶連接
        GroupUser.TryGetValue(msgBody.ToId, out var currentGroup);
        foreach (var user in currentGroup)
        {
            if (user == msgBody.FromId)
            {
                continue;
            }
            //通過群組內的用戶標識去用戶集合獲取用戶集合裡的用戶唯一連接發送消息
            if (UserConnection.TryGetValue(user, out var targetSocket) && targetSocket.State == WebSocketState.Open)
            {
                await targetSocket.SendAsync(new ArraySegment<byte>(sendByte, 0, sendByte.Length), WebSocketMessageType.Text, true, CancellationToken.None);
            }
            else
            {
                currentGroup.Remove(user);
            }
        }
    });
    _disposables.TryAdd(channel, sub);
}

全員消息處理

全員消息處理相對來說思路比較簡單，因為當服務啟動的時候就會監聽redis的全員消息頻道，這樣的話具體的實現也就隻包含發送和接收全員消息瞭，首先看一下全員消息發送的邏輯

private async Task HandleAll(string id, object msgBody)
{
    _logger.LogInformation($"user {id} send:{msgBody}");
    //直接給redis的全員頻道發送消息
    ChannelMsgBody channelMsgBody = new ChannelMsgBody { FromId = id, Msg = msgBody.ToString() };
    _redisClient.Publish(all, JsonConvert.SerializeObject(channelMsgBody));
}

全員消息的發送數據格式如下所示

{"Method":"All", "MsgBody":"Hello All"}

Method為All代表著全員消息類型，MsgBody則代表著具體消息。接收消息出裡同樣很簡單，訂閱redis全員消息頻道，然後遍歷當前WebSocket服務器實例內的所有用戶獲取連接發送消息，具體邏輯如下

private async Task SubAllMsg(string channel)
{
    var sub = _redisClient.Subscribe(channel, async (channel, data) =>
    {
        ChannelMsgBody msgBody = JsonConvert.DeserializeObject<ChannelMsgBody>(data.ToString());
        byte[] sendByte = Encoding.UTF8.GetBytes($"user 【{msgBody.FromId}】 send all:{msgBody.Msg}");
        //獲取當前服務器實例內所有用戶的連接
        foreach (var user in UserConnection)
        {
            //不給自己發送消息，因為發送的時候可以通過具體的業務代碼處理
            if (user.Key == msgBody.FromId)
            {
                continue;
            }
            //給每個用戶發送消息
            if (user.Value.State == WebSocketState.Open)
            {
                await user.Value.SendAsync(new ArraySegment<byte>(sendByte, 0, sendByte.Length), WebSocketMessageType.Text, true, CancellationToken.None);
            }
            else
            {
                _ = UserConnection.TryRemove(user.Key, out _);
            }
        }
    });
    _disposables.TryAdd(channel, sub);
}

示例源碼

由於篇幅有限，沒辦法設計到全部的相關源碼，因此在這裡貼出來github相關的地址，方便大傢查看和運行源碼。相關的源碼我這裡實現瞭兩個版本，一個是基於asp.net core的版本，一個是基於golang的版本。兩份源碼的實現思路是一致的，所以這兩份代碼可以運行在一套集群示例裡，配置在一套nginx裡，並且連接到同一個redis實例裡即可

asp.net core源碼示例https://github.com/softlgl/WebsocketCluster
golang源碼示例https://github.com/softlgl/websocket-cluster

倉庫裡還涉及到本人閑暇之餘開源的其他倉庫，由於本人能力有限難登大雅之堂，就不做廣告瞭，有興趣的同學可以自行瀏覽一下。

總結

本文基於ASP.NET Core框架提供瞭一個基於WebSocket做集群的示例，由於思想是通用的，所以基於這個思路樓主也實現瞭golang版本。其實在之前就想自己動手搞一搞關於WebSocket集群方面的設計，本篇文章算是對之前想法的一個落地操作。其核心思路文章已經做瞭相關介紹，由於這些隻是博主關於構思的實現，可能有很多細節尚未體現到，還希望大傢多多理解。其核心思路總結一下

首先是，利用可以構建WebSocket服務的框架，在當前服務實例中保存當前客戶端用戶和WebSocket的連接關系
如果消息的目標客戶端不在當前服務器，可以利用redis頻道、消息隊列相關、甚至是數據庫類的共享回話發送的消息，由目標服務器獲取目標是否屬於自己的ws會話
本文設計的思路使用的是無狀態的方式，即WebSocket服務實例之間不存在直接的消息通信和相互的服務地址存儲，當然也可以利用redis等存儲在線用戶信息等，這個可以參考具體業務自行設計

讀萬卷書,行萬裡路。在這個時刻都在變化點的環境裡，唯有不斷的進化自己，多接觸多嘗試不用的事物，多擴展自己的認知思維，方能構建自己的底層邏輯。畢竟越底層越抽象，越通用越抽象。面對未知的挑戰，自身作為自己堅強的後盾，可能才會讓自己更踏實。

以上就是ASP.NET Core WebSocket集群實現思路詳解的詳細內容，更多關於ASP.NET Core WebSocket的資料請關註WalkonNet其它相關文章！

ASP.NET Core WebSocket集群實現思路詳解

目錄

前言

實現

nginx配置

一對一發送

群組發送

發送所有人

整合到一起

一對一處理

群組處理

全員消息處理

示例源碼

總結

推薦閱讀：

發佈留言取消回覆

近期文章

目錄

前言

實現

nginx配置

一對一發送

群組發送

發送所有人

整合到一起

一對一處理

群組處理

全員消息處理

示例源碼

總結

推薦閱讀：

發佈留言 取消回覆

近期文章

標籤

發佈留言取消回覆