RxJS WebSocket重连后如何重新订阅事件？

你遇到的问题核心是：旧WebSocket连接上的订阅绑定在已断开的实例上，重连后的新连接和这些订阅没有关联，所以即便收到消息也触发不了回调。下面给两种可行的解决方式：

核心思路是把WebSocket连接封装成一个持续的流，所有业务事件的订阅都基于这个流。当连接重连时，流会自动切换到新的WebSocket实例，订阅就能自动生效，无需手动干预。

示例代码（Angular WebSocket服务） #

import { Injectable } from '@angular/core';
import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
import { retry, filter, map } from 'rxjs/operators';

@Injectable({ providedIn: 'root' })
export class WebSocketService {
  private ws$: WebSocketSubject<any>;

  constructor() {
    // 初始化带重连逻辑的WebSocket流
    this.ws$ = webSocket({
      url: 'wss://your-websocket-url',
      openObserver: {
        next: () => {
          console.log('WebSocket连接/重连成功');
          this.sendHeartbeat();
        }
      },
      closeObserver: {
        next: () => {
          console.log('WebSocket连接断开');
        }
      }
    }).pipe(
      // 重连延迟策略，可根据需求调整
      retry({ delay: 3000 })
    );
  }

  // 心跳逻辑（你已实现的部分）
  private sendHeartbeat() {
    const heartbeatInterval = setInterval(() => {
      if (this.ws$.closed) {
        clearInterval(heartbeatInterval);
        return;
      }
      this.ws$.next({ type: 'heartbeat' });
    }, 10000);
  }

  // 通用事件订阅方法
  onEvent<T>(eventType: string) {
    return this.ws$.pipe(
      filter(msg => msg.type === eventType),
      map(msg => msg.data as T)
    );
  }

  // 发送消息方法
  sendMessage(msg: any) {
    this.ws$.next(msg);
  }
}

组件中使用 #

import { Component, OnInit } from '@angular/core';
import { WebSocketService } from './web-socket.service';

@Component({
  selector: 'app-profile',
  templateUrl: './profile.component.html'
})
export class ProfileComponent implements OnInit {
  constructor(private wsService: WebSocketService) {}

  ngOnInit() {
    // 订阅editProfile事件，重连后自动生效
    this.wsService.onEvent('editProfile').subscribe(data => {
      console.log('收到editProfile消息:', data);
      // 你的业务逻辑处理
    });
  }
}

这个方案的优势在于：所有业务订阅都绑定在持续的WebSocket流上，重连后流会自动切换到新连接，订阅者无需任何额外操作就能接收新消息。

如果现有架构不想大幅改动，可以在服务里暴露一个重连成功的通知流，组件监听该流后手动触发重新订阅逻辑。

示例代码（WebSocket服务） #

import { Injectable } from '@angular/core';
import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
import { retry, Subject } from 'rxjs';

@Injectable({ providedIn: 'root' })
export class WebSocketService {
  private ws$: WebSocketSubject<any>;
  // 暴露重连成功的通知Subject
  public reconnected$ = new Subject<void>();

  constructor() {
    this.initWebSocket();
  }

  private initWebSocket() {
    this.ws$ = webSocket({
      url: 'wss://your-websocket-url',
      openObserver: {
        next: () => {
          console.log('WebSocket连接/重连成功');
          // 重连成功时发送通知
          this.reconnected$.next();
          this.sendHeartbeat();
        }
      },
      closeObserver: {
        next: () => {
          console.log('WebSocket连接断开');
        }
      }
    }).pipe(
      retry({ delay: 3000 })
    );
  }

  // 原有事件订阅方法
  subscribeToEvent(eventType: string, callback: (data: any) => void) {
    return this.ws$.pipe(filter(msg => msg.type === eventType)).subscribe(callback);
  }

  // 心跳逻辑（你已实现的部分）
  private sendHeartbeat() {
    // ...你的心跳代码
  }
}

组件中使用 #

import { Component, OnInit, OnDestroy } from '@angular/core';
import { WebSocketService } from './web-socket.service';
import { Subscription } from 'rxjs';

@Component({
  selector: 'app-profile',
  templateUrl: './profile.component.html'
})
export class ProfileComponent implements OnInit, OnDestroy {
  private editProfileSub?: Subscription;
  private reconnectedSub?: Subscription;

  constructor(private wsService: WebSocketService) {}

  ngOnInit() {
    // 首次订阅
    this.subscribeToEditProfile();
    // 监听重连事件，重新订阅
    this.reconnectedSub = this.wsService.reconnected$.subscribe(() => {
      this.subscribeToEditProfile();
    });
  }

  private subscribeToEditProfile() {
    // 先取消旧订阅，避免内存泄漏和重复回调
    this.editProfileSub?.unsubscribe();
    this.editProfileSub = this.wsService.subscribeToEvent('editProfile', data => {
      console.log('收到editProfile消息:', data);
      // 你的业务逻辑处理
    });
  }

  ngOnDestroy() {
    this.editProfileSub?.unsubscribe();
    this.reconnectedSub?.unsubscribe();
  }
}

注意：手动订阅时一定要先取消旧订阅，避免内存泄漏和重复触发回调。

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