RxJS实现WebSocket断开后自动重连并重新订阅频道

你的代码里，ws$.next(...)只在初始连接时发送了一次订阅消息，但RxJS的retry操作符重连时会创建新的WebSocket连接，这个新连接不会自动执行之前的next调用——这就是重连后无法重新订阅的核心原因。另外你没有将pipe(retry)后的Observable赋值给变量，实际订阅的还是原始的无重试逻辑的ws$（不过你提到重连已实现，可能实际代码有调整，但核心问题始终是订阅消息只触发了一次）。

要让每次重连后自动发送订阅消息，必须把发送订阅的逻辑绑定到每次连接建立成功的时机上。下面提供两种可行实现方式：

方式一：通过重新创建WebSocket主题实现 #

import WebSocket from 'ws';
import { retry, switchMap, of } from 'rxjs';
import { webSocket } from 'rxjs/webSocket';

// 封装创建WebSocket主题的函数
const createWs = () => webSocket({
  url: '...',
  WebSocketCtor: WebSocket as any
});

// 构建带重连+自动订阅的数据流
const wsStream$ = of(null).pipe(
  switchMap(() => {
    const ws$ = createWs();
    // 新连接建立后立即发送订阅消息
    ws$.next({cmd: 'subscribe', channel: 'updates'});
    return ws$;
  }),
  retry({ delay: 3000 })
);

// 订阅数据流
wsStream$.subscribe({
  next: msg => console.log('msg', msg),
  error: e => console.error('error', e),
  complete: () => console.log('complete')
});

方式二：利用openObserver监听连接事件 #

import WebSocket from 'ws';
import { retry } from 'rxjs/operators';
import { webSocket } from 'rxjs/webSocket';

const ws$ = webSocket({
  url: '...',
  WebSocketCtor: WebSocket as any,
  // 监听连接打开事件，每次连接成功（含重连）都发送订阅消息
  openObserver: {
    next: () => ws$.next({cmd: 'subscribe', channel: 'updates'})
  }
});

// 应用重连逻辑并订阅
ws$.pipe(retry({ delay: 3000 })).subscribe({
  next: msg => console.log('msg', msg),
  error: e => console.error('error', e),
  complete: () => console.log('complete')
});

1. 重连逻辑：retry操作符会在WebSocket连接断开抛出错误时，重新订阅上游Observable。RxJS的webSocket主题在连接断开时会触发错误通知，从而触发retry重新创建连接。
2. 自动订阅的核心：
   • 方式一中，每次重连都会通过switchMap新建WebSocket主题，并立即发送订阅消息，确保每个新连接都执行订阅动作。
   • 方式二中，openObserver会监听WebSocket的open事件，无论首次连接还是重连后的连接，只要成功建立就会触发回调发送订阅消息。
3. 原始代码的问题本质：直接调用ws$.next(...)只会在当前连接实例上执行一次，重连后的新连接是独立实例，不会继承这个操作。必须将订阅动作与连接生命周期绑定，才能保证每次连接都执行订阅。

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