RXJS WebSocket Subject订阅异常：Angular组件重连后无法接收消息

问题原因分析 #

你碰到的这个问题其实是RxJS Subject实例的生命周期导致的：

• 服务里的create()方法每次调用（包括重连时）都会创建全新的WebSocketSubject实例，并覆盖this.wsSubj。
• 但你的组件在ngOnInit里只订阅了第一次调用create()时返回的那个旧Subject。当WebSocket断开触发retryWhen重建连接后，服务内部的wsSubj已经换成新对象了，但组件的订阅还挂在那个已经进入错误/完成状态的旧Subject上，自然收不到后续的消息。

简单说：组件和服务后续的WebSocket连接“失联”了，因为订阅的不是同一个流。

修复方案 #

方案1：用中转Subject统一消息分发（推荐）

这种方式让服务内部维护一个全局的消息中转流，不管WebSocket连接重建多少次，所有消息都会转发到这个中转流，组件只需要订阅一次就能持续接收消息。

修改服务代码：

@Injectable({ providedIn: 'root', })
export class WebSocketChannel {
    // 全局中转流，专门用来转发WebSocket消息
    private _messageStream$ = new Subject<WebSocketPackage<any>>();
    // 暴露给组件的只读Observable
    public message$ = this._messageStream$.asObservable();
    private wsSubject?: WebSocketSubject<WebSocketPackage<any>>;

    create(url?: string): void {
        this.wsSubject = webSocket<WebSocketPackage<any>>(this.channelUrl);
        this.wsSubject.pipe(
            retryWhen(errors => errors.pipe(
                tap(err => console.error('Websocket error', err)),
                delay(1000)
            )),
            // 把所有WebSocket消息转发到中转流
            tap(message => this._messageStream$.next(message))
        ).subscribe(
            () => console.log('ok'),
            () => console.log('ko')
        );
    }
}

组件代码修改：

ngOnInit() {
    this.wsManager.create();
    // 只需要订阅服务暴露的message$即可
    this.wsManager.message$
        .subscribe(
            data => console.log('HomeComponent WS OK'),
            err => console.log('HomeComponent WS ERROR CATCH'),
            () => console.log('complete')
        );
}

方案2：让服务返回带重试逻辑的持久化流

这种方式把重连逻辑直接整合到返回的Observable中，组件订阅后，流会自动处理重连，不需要关心内部的Subject替换。

修改服务代码：

@Injectable({ providedIn: 'root', })
export class WebSocketChannel {
    create(url?: string): Observable<WebSocketPackage<any>> {
        // 直接返回带有重试逻辑的WebSocket流
        return webSocket<WebSocketPackage<any>>(this.channelUrl).pipe(
            retryWhen(errors => errors.pipe(
                tap(err => console.error('Websocket error', err)),
                delay(1000)
            )),
            tap(() => console.log('ok'))
        );
    }
}

组件代码修改：

ngOnInit() {
    // 直接订阅create()返回的流，它会自动处理重连
    this.wsManager.create()
        .subscribe(
            data => console.log('HomeComponent WS OK'),
            err => console.log('HomeComponent WS ERROR CATCH'),
            () => console.log('complete')
        );
}

方案对比 #

• 方案1更适合需要在服务内部做额外消息处理（比如过滤、转换），或者多个组件需要共享WebSocket消息的场景。
• 方案2更简洁，符合RxJS的“流优先”思维，适合单组件使用WebSocket的场景。

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