IdentityServer4 + SignalR Core +RabbitMQ 構建web即時通訊（三）

IdentityServer4 + SignalR Core +RabbitMQ 構建web即時通訊（三）

---

 

后台服務用戶與認證 

---

 

新建一個空的.net core web項目Demo.Chat，端口配置為5001，安裝以下nuget包

1.IdentityServer4.AccessTokenValidation，IdentityServer4客戶端認證所用；

2.Microsoft.AspNetCore.SignalR

3.RabbitMQ.Client

添加appsettings.json

{
  "RabbitMQ": {
    "Host": "192.168.1.107",
    "User": "admin",
    "Password": "123123"
  },
  "Authentication": {
    "Authority": "http://localhost:5000"
  }
}

這里我們新增兩個Dto類，一個消息傳輸類MsgDto，一個用戶數據類UserDto

    public class MsgDto
    {
        public UserDto FromUser { get; set; }
        public UserDto ToUser { get; set; }
        public string Content { get; set; }
        public DateTime SendTime { get; set; }
    }
    public class UserDto
    {
        // signalr當前的連接id
        public string ConnectionId { get; set; }
        public Guid Id { get; set; }
        public string UserName { get; set; }
        public string EMail { get; set; }
        public string Avatar { get; set; }
    }

當用戶認證通過后，從Identity返回的token中我們已經返回了用戶的基礎信息了，那這里我們如何獲取呢？很簡單在上下文的User中Claims屬性里面，所以這里我們增加一個擴展方法來轉換為UserDto 

        public static UserDto GetUser(this ClaimsPrincipal claimsPrincipal)
        {
            return new UserDto
            {
                Id = new Guid(claimsPrincipal.Claims.FirstOrDefault(r => r.Type == "sub").Value),
                EMail = claimsPrincipal.Claims.FirstOrDefault(r => r.Type == "email").Value,
                UserName = claimsPrincipal.Claims.FirstOrDefault(r => r.Type == "username").Value,
                Avatar = claimsPrincipal.Claims.FirstOrDefault(r => r.Type == "avatar").Value,
            };
        }

既然是在線聊天那必須得存儲當前所有的在線用戶對吧？新建一個OnlineUsers類，這里我們就不用數據庫了，Demo嘛，里面就3個用戶，嘿嘿。當然你完全可以自由發揮使用其他redis，mongo什么什么的。

    public class OnlineUsers
    {
        /// <summary>
        /// 用戶id作為key
        /// </summary>
        private static ConcurrentDictionary<Guid, UserDto> onlineUsers { get; } = new ConcurrentDictionary<Guid, UserDto>();

        public void AddOrUpdateUser(UserDto user)
        {
            onlineUsers.AddOrUpdate(user.Id, user, (id, r) => user);
        }

        public List<UserDto> Get()
        {
            return onlineUsers.Values.ToList();
        }

        public UserDto Get(Guid userId)
        {
            onlineUsers.TryGetValue(userId, out UserDto user);
            return user;
        }

        public void Remove(Guid userId)
        {
            if (onlineUsers.ContainsKey(userId))
                onlineUsers.TryRemove(userId, out UserDto user);
        }
    }

 

后台服務RabbitMQ消息處理

---

 

RabbitMQ消息隊列相關的知識這里我也不再贅述，園子里面很多，大家自行研究，RabbitMQ大概有2個種模式：生產消費者模式和發布/訂閱模式，生產消費者模式即消息只能被使用一次，比如一個商品生產出來你只能賣給一個消費者對吧，發布/訂閱即只要訂閱了都會收到該消息。這里我們用到的是生產消費者模式，參考官方文檔。

消息發送和收到消息的處理，這里我們分為2個類單獨處理，MsgSender和MsgHandler。

MsgSender:當用戶發送了一條消息，后端收到后就將消息添加到消息隊列，MsgHandler：一直處於運行狀態，當收到隊列的消息時，開始處理消息，調用SignalR的方法，發送消息到客戶端，RabbitMQ的連接配置在appsettings.json中，注入IConfiguration獲取

MsgSender

    public class MsgSender
    {
        public MsgSender(IConfiguration configuration)
        {
            factory = new ConnectionFactory();
            factory.HostName = configuration.GetValue<string>("RabbitMQ:Host");
            factory.UserName = configuration.GetValue<string>("RabbitMQ:User");
            factory.Password = configuration.GetValue<string>("RabbitMQ:Password");
        }
        ConnectionFactory factory;

        public void Send(MsgDto msg)
        {
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare("chat_queue", false, false, false, null);//創建一個名稱為hello的消息隊列
                    var body = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(msg));
                    channel.BasicPublish("", "chat_queue", null, body); //開始傳遞
                }
            }
        }
    }

MsgHandler，需要注入IHubContext接口，用於發送消息到客戶端,ps:在Hub類中,可以通過Clients直接發送消息到客戶端,在其他類里面可以使用這個接口,獲取到Clients。

    public class MsgHandler : IDisposable
    {
        public MsgHandler(IConfiguration configuration, IHubContext<MessageHub> hubContext)
        {
            factory = new ConnectionFactory();
            factory.HostName = configuration.GetValue<string>("RabbitMQ:Host");
            factory.UserName = configuration.GetValue<string>("RabbitMQ:User");
            factory.Password = configuration.GetValue<string>("RabbitMQ:Password");
            this.hubContext = hubContext;
            connection = factory.CreateConnection();
            channel = connection.CreateModel();

        }
        ConnectionFactory factory;
        // 注入SignalR的消息處理器上下文，用以發送消息到客戶端
        IHubContext<MessageHub> hubContext;
        IConnection connection;
        IModel channel;
        public void BeginHandleMsg()
        {
            channel.QueueDeclare("chat_queue", false, false, false, null);
            var consumer = new EventingBasicConsumer(channel);
            channel.BasicConsume("chat_queue", false, consumer);
            consumer.Received += (model, arg) =>
            {
                var body = arg.Body;
                var message = Encoding.UTF8.GetString(body);
                var msg = JsonConvert.DeserializeObject<MsgDto>(message);
                // 通過消息處理器上下文發送消息到客戶端
                hubContext.Clients?.Client(msg.ToUser.ConnectionId)
                                  ?.SendAsync("Receive", msg);

                channel.BasicAck(arg.DeliveryTag, false);
            };
        }

        public void Dispose()
        {
            channel?.Dispose();
            connection?.Dispose();
        }
    }

 

后台服務SignalR消息處理器

---

 

關於SignalR，官方文檔

SignalR的核心就是繼承自Hub消息處理類，這個類中所有的public 方法都可以給客戶端調用。我們的聊天室比較簡陋，只需要一個Send方法給客戶端就夠了，是吧？當然服務端需要2個主動發送消息到客戶端的方法，1.當有用戶登錄時通知所有客戶端刷新在線用戶列表，2.有什么錯誤的時候發送錯誤消息給客戶端，比如我們不允許離線發送，用戶發了條消息給一個不在線的用戶。

另外當用戶登錄和離開時需要在OnlineUsers中進行注冊和注銷。

MessageHub，我們的聊天室必須登錄，所以加上Authorize特性。

    [Authorize]
    public class MessageHub : Hub
    {
        MsgSender msgSender;
        MsgHandler msgQueueHandler;
        OnlineUsers onlineUsers;
        public MessageHub(MsgSender msgSender, MsgHandler msgQueueHandler, OnlineUsers onlineUsers)
        {
            this.msgSender = msgSender;
            this.msgQueueHandler = msgQueueHandler;
            this.onlineUsers = onlineUsers;
        }

        public async Task Send(string toUserId, string message)
        {
            string timestamp = DateTime.Now.ToShortTimeString();
            var toUser = onlineUsers.Get(new Guid(toUserId));
            if (toUser == null)
            {
                await SendErrorAsync("用戶已離線");
                return;
            }
            var fromUser = Context.User.GetUser();
            msgSender.Send(new Dtos.MsgDto
            {
                Content = message,
                FromUser = fromUser,
                SendTime = DateTime.Now,
                ToUser = toUser
            });
        }

        /// <summary>
        /// 當有用戶登錄時 添加在線用戶，並設置用戶的ConnectionId
        /// </summary>
        /// <returns></returns>
        public override async Task OnConnectedAsync()
        {
            await base.OnConnectedAsync();
            var user = Context.User.GetUser();
            if (user == null)
            {
                await SendErrorAsync("您沒有登錄");
                return;
            }
            user.ConnectionId = Context.ConnectionId;
            onlineUsers.AddOrUpdateUser(user);
            await SendUserInfo();
            await RefreshUsersAsync();
        }

        /// <summary>
        /// 當有用戶離開時，注銷用戶登錄
        /// </summary>
        /// <param name="exception"></param>
        /// <returns></returns>
        public override async Task OnDisconnectedAsync(Exception exception)
        {
            //disconnection
            await base.OnDisconnectedAsync(exception);
            var userId = Context.User?.GetUser()?.Id;
            if (userId.HasValue)
                onlineUsers.Remove(userId.Value);
            await RefreshUsersAsync();
        }

        private async Task RefreshUsersAsync()
        {
            var users = onlineUsers.Get().Where(r => r.Id != Context.User.GetUser().Id).ToList();
            // 發送給所有的在線客戶端，通知刷新在線用戶
            await Clients.All.SendAsync("Refresh", users);
        }

        private async Task SendErrorAsync(string errorMsg)
        {
            // 發送錯誤消息給調用者
            await Clients.Caller.SendAsync("Error", errorMsg);
        }

    }

 

這里就冒出來另外一個新的問題了，SignalR使用的是websocket，據我了解到的是沒有header頭這個東西的，而jwt token默認是通過header中Authorization信息進行認證的。那這個授權又如何實現呢？想辦法咯，既然header傳不進來，那直接url傳進來總可以吧。

 

后台服務：服務注冊與認證授權

---

 

好了，我們先將需要的服務先配置下。

AddIdentityServerAuthentication實際上是AddJwtBearer的擴展，你要喜歡也可以用AddJwtBearer配置，由IdentityServer4.AccessTokenValidation提供，配置認證Authority為http://localshot:5000(Demo.Identity配置的端口號為5000，appsetting.json中配置),ApiName和Secret與Identity端配置的ApiResource一致。

        public void ConfigureServices(IServiceCollection services)
        {
            // 注冊消息處理器 消息發送器，在線用戶類
            services.AddSingleton<MsgHandler>()
                .AddSingleton<MsgSender>()
                .AddSingleton<OnlineUsers>();

            // 增加認證服務
            services.AddAuthentication(r =>
            {
                r.DefaultScheme = "JwtBearer";
            })
            // 增加jwt認證
            .AddIdentityServerAuthentication("JwtBearer", r =>
            {
                // 配置認證服務器
                r.Authority = Configuration.GetValue<string>("Authentication:Authority");
                // 配置無需驗證https
                r.RequireHttpsMetadata = false;
                // 配置 當前資源服務器的名稱
                r.ApiName = "chatapi";
                // 配置 當前資源服務器的連接密碼
                r.ApiSecret = "123123";
                r.SaveToken = true;
            });

            // 跨域
            services.AddCors(r =>
            {
                r.AddPolicy("all", policy =>
                {
                    policy
                    .AllowAnyOrigin()
                    .AllowAnyHeader()
                    .AllowAnyMethod()
                    .AllowCredentials()
                    ;
                });
            });
            // 增加授權服務
            services.AddAuthorization();
            // 增加SignalR 服務
            services.AddSignalR();
        }

剛剛提到SignalR認證的問題，具體如何實現呢？這里也有2種方式，1.使用中間件在認證之前從url中獲取token並添加到header中；2.r.MapHub<MessageHub>("/msg")，可以配置在參數中添加自定義的IAuthorizeData接口，可以自己實現獲取token驗證，我覺得比較麻煩，這里我們使用第一種方式。

添加中間件，這個中間件一定要在UseAuthentication之前：

            // signalr jwt認證 token添加
            app.Use(async (context, next) =>
            {
                // 這里從url中獲取token參數，實際應用請實際考慮，加一些過濾條件
                if (context.Request.Query.TryGetValue("token", out var token))
                {
                    // 從url中拿到header，再添加到header中，一定要在UseAuthentication之前
                    context.Request.Headers.Add("Authorization", $"Bearer {token}");
                }
                await next.Invoke();
            });

好了，還有一個問題，前面寫的MsgHandler什么時候開始處理消息？Dispose什么時候調用？這里我們使用IApplicationLifetime接口，該接口提供了應用的整個生命周期事件處理。在應用啟動的時候我們注冊消息處理，應用結束時Dispose。

            // 應用啟動時開始處理消息
            applicationLifetime.ApplicationStarted.Register(msgHandler.BeginHandleMsg);
            // 應用退出時,釋放資源
            applicationLifetime.ApplicationStopping.Register(msgHandler.Dispose);

 

完整的Configure代碼：

        public void Configure(
            IApplicationBuilder app,
            IHostingEnvironment env,
            MsgHandler msgHandler,
            IApplicationLifetime applicationLifetime)
        {
            if (env.IsDevelopment())
            {
                app.UseDeveloperExceptionPage();
            }
            app.UseStaticFiles();
            app.UseCors("all");

            app.UseAuthentication();
            // 使用SignalR 並添加MessageHub類的消息處理器
            app.UseSignalR(r =>
            {
                r.MapHub<MessageHub>("/msg");
            });           

            // 應用啟動時開始處理消息
            applicationLifetime.ApplicationStarted.Register(msgHandler.BeginHandleMsg);
            // 應用退出時,釋放資源
            applicationLifetime.ApplicationStopping.Register(msgHandler.Dispose);
        }

另外用戶登錄后需要展示用戶信息，郵件地址啊頭像什么的，這里我們也有2種方式，1是消息處理器中，當用戶連接后主動發送消息給用戶；2是建一個Api接口，當然放在消息處理器中會顯得更純潔，web項目里面沒有一個controller，這里我們使用第一種方式。

在MessageHub中添加方法，在OnConnectedAsync方法中調用

        private async Task SendUserInfo()
        {
            await Clients.Caller.SendAsync("UserInfo", Context.User.GetUser());
        }

 

聊天室web前端

---

 

官方提供了js庫，可以用npm安裝，npm install @aspnet/signalr。

這個前端嘛，我就不花大功夫去做得漂亮高大上了，暫時就把代碼直接丟在Demo.chat里面吧,2個頁面，登錄頁login，聊天室頁面chat。

 

 關於前端就不啰嗦了，再啰嗦就是關公面前耍大刀了，什么angular,vue,老夫寫代碼統統jquery。其他的大家自己發揮了。

login.html

<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8" />
    <title>登錄聊天室</title>
    <script src="https://cdn.bootcss.com/jquery/3.3.1/jquery.min.js"></script>
</head>
<body>
    <div>
        <fieldset>
            <legend>登錄聊天室</legend>
            <div>
                <input type="text" name="uername" id="username" value="" />
            </div>
            <div>
                <input type="password" name="password" id="password" value="" />
            </div>
            <div>
                <button id="login" type="button">登錄</button>
            </div>
        </fieldset>
    </div>
    <script type="text/javascript">
        $(function () {
            var identityUrl = 'http://localhost:5000/connect/token';

            $('#login').click(function () {

                $.post(identityUrl, {
                    client_id: 'chat_client',
                    grant_type: 'password',
                    scope: 'openid chatapi profile offline_access',
                    username: $('#username').val(),
                    password: $('#password').val()
                }, function (result) {
                    if (result && result.access_token) {
                        sessionStorage['token'] = result.access_token;
                        window.location = "http://localhost:5001/chat.html";
                    }
                }, 'json');
            });
        });
    </script>
</body>
</html>

chat.html

<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8" />
    <title></title>
    <script src="https://cdn.bootcss.com/jquery/3.3.1/jquery.min.js"></script>
    <script src="/lib/signalr.js"></script>
</head>
<body>
    <div>
        <div>
            <input type="hidden" id="userId" value="" />
            <p><label>UserName:</label><i id="userName"></i></p>
            <p><label>EMail:</label><i id="email"></i></p>
            <p><label>Avatar:</label><img id="avatar" style="width:48px; height:48px; border-radius:50%; overflow:hidden;" /> </p>
        </div>
        <div style="width:700px;height:500px;border:1px solid red;">
            <ul id="msgList"></ul>
        </div>
        <div>
            <select id="users"></select>
        </div>
        <div>
            <textarea id="msgSendContent" placeholder="請輸入發送消息" cols="100" rows="4"></textarea>
            <br />
            <button id="send" type="button">發送</button>
        </div>

    </div>

    <script type="text/javascript">
        $(function () {
            var token = sessionStorage['token'];

            if (!token) {
                alert('請先登錄!');
                window.location = 'http://localhost:5001/login.html';
                return;
            }
            function timeFormat(time) {
                time = new Date(time)
                return time.toLocaleDateString() + ' ' + time.toLocaleTimeString();
            }

            var connection = new signalR.HubConnectionBuilder()
                .withUrl("/msg?token=" + token)
                .configureLogging(signalR.LogLevel.Information)
                .build();

            connection.on('Receive', function (msg) {
                var $ul = $('#msgList');
                var $li = $('<li>' + msg.fromUser.userName + '[' + timeFormat(msg.sendTime) + '] : ' + msg.content + '</li>');
                $ul.append($li);
            });

            connection.on('UserInfo', function (userInfo) {
                $('#userName').text(userInfo.userName);
                $('#email').text(userInfo.eMail);
                $('#avatar').attr('src', userInfo.avatar);
                $('#userId').val(userInfo.id);
            });

            connection.on('Refresh', function (users) {
                $('#users').empty();
                users.forEach(function (user) {
                    if (user.id != $('#userId').val())
                        $('#users').append('<option value="' + user.id + '">' + user.userName + '</option>');
                });;
            });

            connection.on('Error', function (err) {
                alert(err);
            });

            connection.start().catch(err => console.error(err.toString()));
            $('#send').click(function () {
                var msg = $('#msgSendContent').val();
                var toUerId = $('#users').val();
                connection.invoke('Send', toUerId, msg).catch(err => console.error(err));
                var $ul = $('#msgList');
                var $li = $('<li>我[' + timeFormat(new Date()) + '] : ' + msg + '</li>');
                $ul.append($li);
            });
        });
    </script>
</body>
</html>

 

好了，代碼就寫完了，同時運行Demo.Identity和Demo.Chat。打開2個瀏覽器：http://localhost:5001/login.html。

 

 輸入用戶名密碼登錄；

發送個消息試試：

是不是很簡陋？嘿嘿

好了，到處為止。其他不完善的地方,自己動手,豐衣足食,如離線消息,token自動刷新等等.