前言
????提到WebSocket相信大家都聽說過，它的初衷是為了解決客戶端瀏覽器與服務端進行雙向通信，是在單個TCP連接上進行全雙工通訊的協議。在沒有WebSocket之前只能通過瀏覽器到服務端的請求應答模式比如輪詢，來實現服務端的變更響應到客戶端，現在服務端也可以主動發送數據到客戶端瀏覽器。WebSocket協議和Http協議平行，都屬于TCP/IP四層模型中的第四層應用層。由于WebSocket握手階段采用HTTP協議,所以也需要進行跨域處理。它的協議標識是ws或wss對應了常規標識和安全通信協議標識。本文重點并不是介紹WebSocket協議相關，而是提供一種基于ASP.NET Core原生WebSocket的方式實現集群的實現思路。關于這套思路其實很早之前我就構思過了，只是之前一直沒有系統的整理出來，本篇文章就來和大家分享一下，由于主要是提供一種思路，所以涉及到具體細節或者業務相關的可能沒有體現出來，還望大家理解。
實現
咱們的重點關鍵字就是兩個WebSocket和集群，實現的框架便是基于ASP.NET Core,我也基于golang實現了一套,本文涉及到的相關源碼和golang版本的實現都已上傳至我的github[1],具體倉庫地址可以轉到文末自行跳轉到示例源碼[2]中查看。既然涉及到集群，這里咱們就用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?群員容器為valuevar?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源碼示例?WebsocketCluster[3]??
golang源碼示例?websocket-cluster[4]
倉庫里還涉及到本人閑暇之余開源的其他倉庫，由于本人能力有限難登大雅之堂，就不做廣告了，有興趣的同學可以自行瀏覽一下。
總結
????本文基于ASP.NET Core框架提供了一個基于WebSocket做集群的示例，由于思想是通用的，所以基于這個思路樓主也實現了golang版本。其實在之前就想自己動手搞一搞關于WebSocket集群方面的設計，本篇文章算是對之前想法的一個落地操作。其核心思路文章已經做了相關介紹，由于這些只是博主關于構思的實現，可能有很多細節尚未體現到，還希望大家多多理解。其核心思路總結一下
??首先是，利用可以構建WebSocket服務的框架，在當前服務實例中保存當前客戶端用戶和WebSocket的連接關系
??如果消息的目標客戶端不在當前服務器，可以利用redis頻道、消息隊列相關、甚至是數據庫類的共享回話發送的消息，由目標服務器獲取目標是否屬于自己的ws會話
??本文設計的思路使用的是無狀態的方式，即WebSocket服務實例之間不存在直接的消息通信和相互的服務地址存儲，當然也可以利用redis等存儲在線用戶信息等，這個可以參考具體業務自行設計
讀萬卷書,行萬里路。在這個時刻都在變化點的環境里，唯有不斷的進化自己，多接觸多嘗試不用的事物，多擴展自己的認知思維，方能構建自己的底層邏輯。畢竟越底層越抽象，越通用越抽象。面對未知的挑戰，自身作為自己堅強的后盾，可能才會讓自己更踏實。
引用鏈接
[1]?我的github:?https://github.com/softlgl[2]?示例源碼:?#示例源碼[3]?WebsocketCluster:?https://github.com/softlgl/WebsocketCluster[4]?websocket-cluster:?https://github.com/softlgl/websocket-cluster
 JVM架構解釋)
