Angular集成SignalR聊天功能 目标客户端不显示新消息

问题现象 #

基于SignalR搭建的实时聊天系统出现如下异常：

• 新消息发送至服务端后，发送方浏览器可正常展示全部新旧消息
• 消息目标客户端的消息框无法加载显示新消息
• 若在NewMessage回调中直接执行this.messageThreadSource.next([message])，会触发两类异常：
  • 目标客户端消息框被清空，所有历史消息从HTML页面移除
  • 发送方客户端消息框仅展示最新发送的单条消息
• 调试确认：NewMessage集线器回调中，message（新消息对象）、messages（已有历史消息数组）两个参数均可正常获取值

原始实现代码 #

baseUrl = environment.apiUrl;
hubUrl = environment.hubUrl;
private hubConnection: HubConnection;
private messageThreadSource = new BehaviorSubject<Message[]>([]);
messageThread$ = this.messageThreadSource.asObservable();

constructor(private http: HttpClient) { }

createHubConnection(userToken: UserToken, otherUserId: string, skip: string) {
  this.hubConnection = new HubConnectionBuilder()
    .withUrl(this.hubUrl + 'message?targetUserId=' + otherUserId + '&&skip=' + skip, {
      accessTokenFactory: () => userToken.token
    })
    .withAutomaticReconnect()
    .build()

  this.hubConnection
    .start()
    .catch(error => console.log("SignalR Message Error : ", error));

  this.hubConnection.on("ReceiveMessageThread", messages => {
    this.messageThreadSource.next(messages);
  });

  this.hubConnection.on("NewMessage", message => {
    this.messageThread$.pipe(take(1)).subscribe(messages => {
      this.messageThreadSource.next([...messages,message]);
    })
  });
}

问题根因 #

1. 事件注册顺序错误：代码先调用start()建立连接，再注册ReceiveMessageThread、NewMessage事件监听。如果连接建立完成瞬间服务端就推送消息，此时监听还未注册，消息会被直接丢弃，目标客户端刚好触发这个时序问题时就收不到新消息更新。
2. 服务端推送逻辑遗漏：服务端处理新消息时，大概率只将NewMessage事件推送给了发送方连接，没有推送给消息接收方对应的在线连接，导致接收方根本不会触发NewMessage回调。
3. 参数拼接错误：Hub连接URL中skip参数前多写了一个&，部分严格校验参数的后端环境会出现参数解析异常，导致初始化加载的消息列表不符合预期。
4. 全量状态覆盖逻辑错误：messageThreadSource是存储全量消息列表的BehaviorSubject，直接传入仅包含单条新消息的数组调用next()，会直接覆盖掉之前存储的所有历史消息，自然会出现列表只剩单条消息、历史被清空的现象。

修复方案 #

1. 调整前端Hub逻辑顺序：先注册所有事件监听，再启动连接，避免启动瞬间丢消息。
2. 修正URL参数拼接错误，删除多余的&。
3. 修改服务端NewMessage推送逻辑：收到新消息后，除了推送给发送方，必须查询消息接收方对应的所有在线连接ID，向这些连接同步推送NewMessage事件。
4. 新消息追加逻辑增加去重判断，避免同一条消息被重复插入列表。

修复后的前端核心代码 #

createHubConnection(userToken: UserToken, otherUserId: string, skip: string) {
  this.hubConnection = new HubConnectionBuilder()
    .withUrl(`${this.hubUrl}message?targetUserId=${otherUserId}&skip=${skip}`, {
      accessTokenFactory: () => userToken.token
    })
    .withAutomaticReconnect()
    .build()

  // 先注册所有事件监听
  this.hubConnection.on("ReceiveMessageThread", messages => {
    this.messageThreadSource.next(messages);
  });

  this.hubConnection.on("NewMessage", message => {
    this.messageThread$.pipe(take(1)).subscribe(messages => {
      // 去重判断：消息已存在则不重复添加
      if (!messages.some(m => m.id === message.id)) {
        this.messageThreadSource.next([...messages, message]);
      }
    })
  });

  // 所有监听注册完成后再启动连接
  this.hubConnection
    .start()
    .catch(error => console.log("SignalR Message Error : ", error));
}

注意：服务端推送时不要仅给当前请求的连接发消息，要根据消息接收方的UserId，从SignalR的连接管理器中查询该用户所有在线连接逐一推送，否则接收方永远收不到新消息事件。

内容的提问来源于stack exchange，提问作者user3748973