关于ASP.NET Core WebSocket实现集群的思考( 四 ) _生活百科

接下来看一下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接收

关于ASP.NET Core WebSocket实现集群的思考( 四 )

经验总结扩展阅读

“好色”的女人,通常都有这3个生活“习惯”,聪明男人一看便知

永劫无间向隅而战活动在哪里

房子赠与和过户的区别及费用有哪些?

失眠小妙招

女人，想要活出“高级感”，请要学会主动丢掉这5样东西！

宝马拿铁三个版本有什么区别？

QQ飞车中的消费卷和点卷有什么区别

汕头金银岛介绍汕头金银岛位置

谁也别想跑第1关怎么过

木材的火灾危险性类别

眉毛变浓的小偏方让你不再担心出门忘画眉

红与黑的作者红与黑小说简介

理发师|理发店理发先洗头再剪头发，里面有什么“套路”？说出来你都不信

林城是哪个城市

星座先森|十二生肖中，一生极容易有好事发生的4个生肖

2022年10月12日乔迁新居吉日一览表 2022年10月12日乔迁新居黄道吉日

女生额前小碎发打理这几个小妙招让你轻松解决烦恼

乐施会|国际组织乐施会：全球最富有的2153人资产比全球46亿人加起来还要多

2022年9月20日适合修造坟墓吗 2022年9月20日是修造坟墓吉日吗

安然姐姐|生活就是酸甜苦辣，一样不差