SignalR实时消息推送

后端

创建一个Hub类，重写连接和断连方法

ChatHub

记录下每一位登录人连接SignalR的信息至Redis中

引用的Microsoft.AspNetCore.App框架

public class ChatHub : Hub
{
    //ConcurrentDictionary<string, HubUser> concurrentDictionary = new ConcurrentDictionary<string, HubUser>();
    private ICurrentLoginInfoService _currentLoginInfoService;
    private IUserSignalRRedisService  _userSignalRRedisService;

    public ChatHub(ICurrentLoginInfoService currentLoginInfoService, IUserSignalRRedisService userSignalRRedisService)
    {
        _currentLoginInfoService = currentLoginInfoService;
        _userSignalRRedisService = userSignalRRedisService;
    }



    /// <summary>
    /// 连接
    /// </summary>
    /// <returns></returns>
    public override async Task OnConnectedAsync()
    {
        //var client = concurrentDictionary.TryGetValue(Context?.ConnectionId, out HubUser? hubUser);
        var client = await _userSignalRRedisService.GetUserConnection(_currentLoginInfoService.UserId);
        if (client == null)
        {
            client = new HubUser()
            {
                ConnectionID = Context.ConnectionId,
                UserId = _currentLoginInfoService.UserId,
                Name = _currentLoginInfoService.UserName
            };
            //concurrentDictionary.GetOrAdd(newUser.ConnectionID, newUser);
        }
        else
        {
            client.ConnectionID = Context.ConnectionId;
        }
        await _userSignalRRedisService.SetUserConnection(_currentLoginInfoService.UserId, client);
        await base.OnConnectedAsync();
    }

    /// <summary>
    /// 断开连接
    /// </summary>
    /// <param name="exception"></param>
    /// <returns></returns>
    public override async Task OnDisconnectedAsync(Exception? exception)
    {
        //var client = concurrentDictionary.TryGetValue(Context.ConnectionId, out HubUser? hubUser);
        var hubUser = await _userSignalRRedisService.GetUserConnection(_currentLoginInfoService.UserId);
        if (hubUser != null)
        {
           await  _userSignalRRedisService.RemoveUserConnection(_currentLoginInfoService.UserId);
            //concurrentDictionary.Remove(hubUser.ConnectionID, out HubUser? reVal);
        }

        await base.OnDisconnectedAsync(exception);
    }

   
}

UserSignalRRedisService

 public class UserSignalRRedisService : IUserSignalRRedisService
 {
     private IRedisServer _redisServer;

     public UserSignalRRedisService(IRedisServer redisServer)
     {
         _redisServer = redisServer;
     }
     /// <summary>
     /// 获取用户的SignalR信息
     /// </summary>
     /// <returns></returns>
     public async Task<HubUser> GetUserConnection(long userId)
     {
         var key = string.Format(GlobalConstants.UserSignalRSession, userId);
         var obj = await _redisServer.GetAsync(key);
         return obj.ToObject<HubUser>();
     }

     /// <summary>
     /// 设置用户的SignalR信息
     /// </summary>
     /// <returns></returns>
     public async Task<bool> SetUserConnection(long userId, HubUser hubUser)
     {
         var key = string.Format(GlobalConstants.UserSignalRSession, userId);
         return await _redisServer.SetAsync(key, hubUser.ToJson());
     }

     /// <summary>
     /// 删除用户的SignalR信息
     /// </summary>
     /// <param name="userId"></param>
     /// <returns></returns>
     public async Task<bool> RemoveUserConnection(long userId)
     {
         return await _redisServer.RemoveAsync(userId.ToString());
     }

 }

ICurrentLoginInfoService

当前登录人信息

 Program.cs引入SignalR

app.MapHub<ChatHub>("/messageHub");

自定义一个中间件来拦截SignalR的请求，做当前登录人信息做处理

public class SignalrInterceptionMiddleware
{
    public const string AuthorizationSchem = "Bearer";
    private readonly RequestDelegate _next;
    private readonly IUserRedisService _redisService;
    private readonly JwtOption _jwtOption;
    private readonly ILogger<SignalrInterceptionMiddleware> _logger;
    public SignalrInterceptionMiddleware(RequestDelegate next, IUserRedisService redisService, IOptions<JwtOption> options, ILogger<SignalrInterceptionMiddleware> logger)
    {
        _next = next;
        _redisService = redisService;
        _jwtOption = options.Value;
        _logger = logger;
    }

    public async Task InvokeAsync(HttpContext context)
    {
        //if(context.WebSockets.IsWebSocketRequest && context.Request.Path.StartsWithSegments("/messageHub"))
        if (context.Request.Path.StartsWithSegments("/messageHub"))
        {
            var accessToken = context.Request.Query["access_token"].ToString();
            if (!string.IsNullOrEmpty(accessToken))
            {
                var token = accessToken.Replace(AuthorizationSchem, string.Empty).Trim();
                var userId = GetUserId(token);
                var redisUser = await _redisService.GetUserSessionAsync(userId);
                if (redisUser == null) throw new CustomException(StatusCode.Unauthorized);
                if (redisUser.Token == null || !redisUser.ExpireDateTime.HasValue) throw new CustomException(StatusCode.Unauthorized);
                if (redisUser.Token != token) throw new CustomException(StatusCode.LoginOnOtherDevice);
                var identity = new ClaimsIdentity(new List<Claim>
                {
                    new Claim(GlobalConstants.UserId,userId.ToString()),
                    new Claim(GlobalConstants.UserName, redisUser.Name.ToString()),

                }, "Custom");
                context.User = new ClaimsPrincipal(identity);
                context.Items.Add(GlobalConstants.UserSession, redisUser);
            }

        }
        await _next(context);
    }
    private long GetUserId(string? token)
    {
        try
        {
            if (token.NotNull())
            {
                var claims = JWTHelper.ValiateToken(token, _jwtOption);
                if (claims != null)
                {
                    var userId = claims.FirstOrDefault(t => t.Type.Equals(GlobalConstants.UserId))?.Value;
                    if (userId.NotNull()) return Convert.ToInt64(userId);
                }
            }

        }
        catch
        {
            _logger.LogError(string.Format("解析token异常，SignalrInterceptionMiddleware=>GetUserId"));
        }
        return default(long);
    }
}

将当前登录人的信息通过token解析到Context.item中方便ICurrentLoginInfoService使用

在 Program.cs中在引入SignalR之前引入该中间件

app.UseMiddleware<SignalrInterceptionMiddleware>();
app.MapHub<ChatHub>("/messageHub");

前端

以vue为例

安装包@microsoft/signalr

npm install @microsoft/signalr

建立一个SignalRHelper的公共类

import * as signalr from '@microsoft/signalr'
export class SingalRHelper {
  connection: any
  retryCount = 0
  maxRetryAttempts = 5
  retryInterval = 5000
  url = `${import.meta.env.VITE_BASEURL}/messageHub`
  constructor() {}
  /**
   * 初始化SignalR
   * @param token
   */
  initSignalR(token: any) {
    this.connection = new signalr.HubConnectionBuilder()
      .withUrl(this.url, {
        skipNegotiation: true,
        transport: signalr.HttpTransportType.WebSockets,
        headers: { Authorization: token },
        accessTokenFactory: () => token
      })
      .build()
    this.startConnection()
  }

  /**
   * 连接
   */
  startConnection() {
    this.connection
      .start()
      .then(() => {
        this.retryCount = 0
        console.log('Connected to SignalR succeddfully!')
      })
      .catch((err: any) => {
        console.log('Error connecting to SignalR:', err)
        if (this.retryCount < this.maxRetryAttempts) {
          setTimeout(() => {
            this.retryCount++
            this.startConnection()
          }, this.retryInterval)
        }
      })
  }

  /**
   * 监听服务端消息
   * @param callback
   */
  onMessageReceived(callback: any) {
    //在连接对象上注册消息接收事件的回调函数
    this.connection.on('MessageReceived', callback)
  }
  /**
   * 断开连接
   */
  stopSignalR() {
    if (this.connection) {
      this.connection.stop()
      console.log('Disconnected from SignalR.')
    }
  }
}

注;SignalR在初始化时Url的目标地址要与后端在Program.cs中引入SignalR的地址保持一致“/messageHub”

页面调用

onMounted(() => {  
   signalR.initSignalR()
   signalR.onMessageReceived((message: any) => {
    console.log('signalR:', message)
  })
})