//上游服務(wù)器地址也就是websocket服務(wù)的真實(shí)地址
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;
        //記得轉(zhuǎn)發(fā)避免踩坑
        proxy_set_header Host $host;
        proxy_http_version 1.1; 
        //http升級成websocket協(xié)議的頭標(biāo)識(shí)
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "Upgrade";
    }
}

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();

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代表當(dāng)前連接的客戶端唯一標(biāo)識(shí)比如用戶唯一標(biāo)識(shí)
    [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();
            //處理請求相關(guān)
            await _socketHandler.Handle(id, webSocket);
        }
        else
        {
            HttpContext.Response.StatusCode = StatusCodes.Status400BadRequest;
        }
    }
}

public class WebSocketHandler:IDisposable
{
    //存儲(chǔ)當(dāng)前服務(wù)用戶的集合
    private readonly UserConnection UserConnection = new();
    //redis頻道前綴
    private readonly string userPrefix = "user:";
    //用戶對應(yīng)的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)
    {
        //把當(dāng)前用戶連接存儲(chǔ)起來
        _ = UserConnection.GetOrAdd(id, webSocket);
        //訂閱一個(gè)當(dāng)前用戶的頻道
        await SubMsg($"{userPrefix}{id}");
        var buffer = new byte[1024 * 4];
        //接收發(fā)送過來的消息，這個(gè)方法是阻塞的，如果沒收到消息則一直阻塞
        var receiveResult = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
        //循環(huán)接收消息
        while (webSocket.State == WebSocketState.Open)
        {
            try
            {
                //因?yàn)榫彌_區(qū)長度是固定的所以要獲取實(shí)際長度
                string msg = Encoding.UTF8.GetString(buffer[..receiveResult.Count]).TrimEnd('\0');
                //接收的到消息轉(zhuǎn)換成實(shí)體
                MsgBody msgBody = JsonConvert.DeserializeObject<MsgBody>(msg);
                //發(fā)送到其他客戶端的數(shù)據(jù)
                byte[] sendByte = Encoding.UTF8.GetBytes($"user {id} send:{msgBody.Msg}");
                _logger.LogInformation($"user {id} send:{msgBody.Msg}");
                //判斷目標(biāo)客戶端是否在當(dāng)前當(dāng)前服務(wù)，如果在當(dāng)前服務(wù)直接扎到目標(biāo)連接直接發(fā)送
                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
                {
                    //如果要發(fā)送的目標(biāo)端不在當(dāng)前服務(wù)，則發(fā)送給目標(biāo)redis端的頻道
                    ChannelMsgBody channelMsgBody = new ChannelMsgBody { FromId = id, ToId = msgBody.Id, Msg = msgBody.Msg };
                    //目標(biāo)的redis頻道
                    _redisClient.Publish($"{userPrefix}{msgBody.Id}", JsonConvert.SerializeObject(channelMsgBody));
                }
                //繼續(xù)阻塞循環(huán)接收消息
                receiveResult = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, ex.Message);
                break;
            }
        }
        //循環(huán)結(jié)束意味著當(dāng)前端已經(jīng)退出
        //從當(dāng)前用戶的集合移除當(dāng)前用戶
        _ = UserConnection.TryRemove(id, out _);
        //關(guān)閉當(dāng)前WebSocket連接
        await webSocket.CloseAsync(receiveResult.CloseStatus.Value, receiveResult.CloseStatusDescription, CancellationToken.None);
        //在當(dāng)前訂閱集合移除當(dāng)前用戶
        _disposables.TryRemove($"{userPrefix}{id}", out var disposable);
        //關(guān)閉當(dāng)前用戶的通道
        disposable.Dispose();
    }
    private async Task SubMsg(string channel)
    {
        //訂閱當(dāng)前用戶頻道
        var sub = _redisClient.Subscribe(channel,  async (channel, data) => {
            //接收過來當(dāng)前頻道數(shù)據(jù)，說明發(fā)送端不在當(dāng)前服務(wù)
            ChannelMsgBody msgBody = JsonConvert.DeserializeObject<ChannelMsgBody>(data.ToString());
            byte[] sendByte = Encoding.UTF8.GetBytes($"user {msgBody.FromId} send:{msgBody.Msg}");
            //在當(dāng)前服務(wù)找到目標(biāo)的WebSocket連接并發(fā)送消息
            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);
    }
    //程序退出的時(shí)候取消當(dāng)前服務(wù)訂閱的redis頻道
    public void Dispose()
    {
        foreach (var disposable in _disposables)
        {
            disposable.Value.Dispose();
        }
        _disposables.Clear();
    }
}

//注冊到當(dāng)前服務(wù)的連接
public class UserConnection : IEnumerable<KeyValuePair<string, WebSocket>>
{
    //存儲(chǔ)用戶唯一標(biāo)識(shí)和WebSocket的對應(yīng)關(guān)系
    private ConcurrentDictionary<string, WebSocket> _users = new ConcurrentDictionary<string, WebSocket>();
    //當(dāng)前服務(wù)的用戶數(shù)量
    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
{
    //目標(biāo)用戶標(biāo)識(shí)
    public string Id { get; set; }
    //要發(fā)送的消息
    public string Msg { get; set; }
}
//頻道訂閱消息
public class ChannelMsgBody
{
    //用戶標(biāo)識(shí)
    public string FromId { get; set; }
    //目標(biāo)用戶標(biāo)識(shí)，也就是要發(fā)送給誰
    public string ToId { get; set; }
    //要發(fā)送的消息
    public string Msg { get; set; }
}

//包含兩個(gè)標(biāo)識(shí)一個(gè)是組別標(biāo)識(shí)一個(gè)是注冊到組別的用戶
[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();
        //調(diào)用HandleGroup處理群組相關(guān)的消息
        await _socketHandler.HandleGroup(groupId, userId, webSocket);
    }
    else
    {
        HttpContext.Response.StatusCode = StatusCodes.Status400BadRequest;
    }
}

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)
    {
        //因?yàn)槿航M的集合可能會(huì)存在很多用戶一起訪問所以限制訪問數(shù)量
        await _lock.WaitAsync();
        //初始化群組容器 群唯一標(biāo)識(shí)為key 群員容器為value
        var currentGroup = GroupUser.Groups.GetOrAdd(groupId, new UserConnection { });
        //當(dāng)前用戶加入當(dāng)前群組
        _ = currentGroup.GetOrAdd(userId, webSocket);
        //只有有當(dāng)前WebSocket服務(wù)的第一個(gè)加入當(dāng)前組的時(shí)候才去訂閱群組頻道
        //如果不限制的話則會(huì)出現(xiàn)如果當(dāng)前WebSocket服務(wù)有多個(gè)用戶在一個(gè)組內(nèi)則會(huì)重復(fù)收到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);
        //服務(wù)不退出的話則一直等待接收
        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頻道發(fā)布的消息，目標(biāo)為群組標(biāo)識(shí)
                ChannelMsgBody channelMsgBody = new ChannelMsgBody { FromId = userId, ToId = groupId, Msg = msg };
                //通過redis發(fā)布消息
                _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;
            }
        }
        //如果客戶端退出則在當(dāng)前群組集合刪除當(dāng)前用戶
        _ = 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}");
            //在當(dāng)前WebSocket服務(wù)器找到當(dāng)前群組里的用戶
            GroupUser.Groups.TryGetValue(msgBody.ToId, out var currentGroup);
            //循環(huán)當(dāng)前WebSocket服務(wù)器里的用戶發(fā)送消息
            foreach (var user in currentGroup)
            {
                //不用給自己發(fā)送了
                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);
                }
            }
        });
        //把當(dāng)前頻道加入訂閱集合
        _disposables.TryAdd(channel, sub);
    }
}

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

//把用戶注冊進(jìn)去
[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;
    }
}

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集群中的每個(gè)服務(wù)只定義一次
        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發(fā)布到集群中的其他服務(wù)
                _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;
            }
        }
        //用戶退出則刪除集合中的當(dāng)前用戶信息
        _ = 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}");
            //接收到消息后遍歷用戶集合把消息發(fā)送給所有用戶
            foreach (var user in AllConnection)
            {   
                //如果包含當(dāng)前用戶跳過
                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);
    }
}

public class WebSocketChannelController : ControllerBase
{
    private readonly ILogger<WebSocketController> _logger;
    private readonly WebSocketChannelHandler _webSocketChannelHandler;
    public WebSocketChannelController(ILogger<WebSocketController> logger, WebSocketChannelHandler webSocketChannelHandler)
    {
        _logger = logger;
        _webSocketChannelHandler = webSocketChannelHandler;
    }
    //只需要把當(dāng)前用戶連接到服務(wù)即可
    [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;
        }
    }
}

public class WebSocketChannelHandler : IDisposable
{
    //用于存儲(chǔ)當(dāng)前WebSocket服務(wù)器鏈接上來的所有用戶對應(yīng)關(guān)系
    private readonly UserConnection UserConnection = new();
    //用于存儲(chǔ)群組和用戶關(guān)系，用戶集合采用HashSet保證每個(gè)用戶只加入一個(gè)群組一次
    private readonly ConcurrentDictionary<string, HashSet<string>> GroupUser = new ConcurrentDictionary<string, HashSet<string>>();
    private readonly SemaphoreSlim _lock = new(1, 1);
    //存放redis訂閱實(shí)例
    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();
        //每次連接進(jìn)來就添加到用戶集合
        _ = UserConnection.GetOrAdd(id, webSocket);
        //每個(gè)WebSocket服務(wù)實(shí)例只需要訂閱一次全員消息頻道
        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);
        //在群組中移除當(dāng)前用戶
        foreach (var users in GroupUser.Values)
        {
            lock (users)
            {
                users.Remove(id);
            }
        }
        //當(dāng)前客戶端用戶退出則移除連接
        _ = 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();
    }
}

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

 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}");
    //判斷目標(biāo)用戶是否在當(dāng)前WebSocket服務(wù)器
    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
    {
        //如果不在當(dāng)前服務(wù)器，則直接把消息發(fā)布到具體的用戶頻道去，由具體用戶去訂閱
        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 _);
            }
        }
    });
    //把訂閱實(shí)例加入集合
    _disposables.TryAdd(channel, sub);
}

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

private async Task AddUserGroup(string user, string group, WebSocket webSocket)
{
    //獲取群組信息
    var currentGroup = GroupUser.GetOrAdd(group, new HashSet<string>());
    lock (currentGroup)
    {
       //把用戶標(biāo)識(shí)加入當(dāng)前組
        _ = currentGroup.Add(user);
    }
    //每個(gè)組的redis頻道，在每臺(tái)WebSocket服務(wù)器實(shí)例只注冊一次訂閱
    if (currentGroup.Count == 1)
    {
        //訂閱當(dāng)前組消息
        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"}

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;
    }
    //只有加入到當(dāng)前群組，才能在群組內(nèi)發(fā)送消息
    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}");
    //發(fā)送群組消息
    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"}

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}");
        //獲取當(dāng)前服務(wù)器實(shí)例中當(dāng)前群組的所有用戶連接
        GroupUser.TryGetValue(msgBody.ToId, out var currentGroup);
        foreach (var user in currentGroup)
        {
            if (user == msgBody.FromId)
            {
                continue;
            }
            //通過群組內(nèi)的用戶標(biāo)識(shí)去用戶集合獲取用戶集合里的用戶唯一連接發(fā)送消息
            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);
}

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

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

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}");
        //獲取當(dāng)前服務(wù)器實(shí)例內(nèi)所有用戶的連接
        foreach (var user in UserConnection)
        {
            //不給自己發(fā)送消息，因?yàn)榘l(fā)送的時(shí)候可以通過具體的業(yè)務(wù)代碼處理
            if (user.Key == msgBody.FromId)
            {
                continue;
            }
            //給每個(gè)用戶發(fā)送消息
            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);
}