如何通过RxJS的WebSocketSubject仅向服务器发送消息？

这问题我之前也碰到过！WebSocketSubject的默认行为确实容易让人踩坑——它本身既是发送消息的Observer，也是接收消息的Observable，所以你调用next()发消息时，它会同时把这条消息推给自己的所有订阅者，相当于“自循环”了。要解决这个问题，核心思路就是把发送消息的通道和接收服务器消息的通道彻底分开，不让本地发送的消息进入订阅流。

给你两种靠谱的解决方案，按优雅程度排序：

方案1：用webSocket()配置的input参数拆分收发流 #

这是RxJS官方推荐的更清晰的写法，直接把发送流和接收流解耦：

import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
import { Subject } from 'rxjs';

// 1. 创建一个专门用于发送消息的Subject
const outgoingMessages$ = new Subject<any>();

// 2. 创建WebSocket连接时，指定input为发送流
const socket$: WebSocketSubject<any> = webSocket({
  url: 'ws://你的服务器地址',
  input: outgoingMessages$ // 这里把发送流传入，负责向服务器发消息
});

// 3. 订阅socket$，这里只会收到服务器返回的消息，完全不会收到自己发的
socket$.subscribe({
  next: (serverMsg) => {
    console.log('收到服务器消息:', serverMsg);
    // 在这里处理业务逻辑就好
  },
  error: (err) => console.error('连接出错:', err),
  complete: () => console.log('连接已关闭')
});

// 4. 发送消息的方法，调用这个就只会发服务器，不会触发本地订阅
const sendToServer = (msg: any) => {
  outgoingMessages$.next(msg);
};

// 示例调用
sendToServer({ type: 'ping', content: 'hello server' });

为什么这个方法管用？因为input参数指定的是仅发送到服务器的数据流，而socket$这个Observable只会推送从服务器接收到的消息，两者完全独立，不会有“自己发的消息自己收”的问题。

方案2：给本地消息加标记，订阅时过滤 #

如果不想改太多现有代码，可以给发送的消息加一个本地标记，然后在订阅流里过滤掉这些标记的消息：

import { WebSocketSubject } from 'rxjs/webSocket';
import { filter } from 'rxjs/operators';

// 创建WebSocketSubject
const socket$ = new WebSocketSubject('ws://你的服务器地址');

// 发送消息时，给消息加一个唯一的本地标记
const sendToServer = (msg: any) => {
  const localMsg = {
    ...msg,
    __isLocal: true // 自定义标记字段，注意不要和服务器返回的字段冲突
  };
  socket$.next(localMsg);
};

// 订阅时过滤掉带本地标记的消息
socket$.pipe(
  filter(msg => !msg.__isLocal)
).subscribe(serverMsg => {
  console.log('收到服务器消息:', serverMsg);
});

这个方法的缺点是需要确保服务器不会返回带有__isLocal字段的消息，否则会误过滤。如果服务器消息格式你能控制，这个方法也能用，但方案1更健壮。

最后提醒下：如果用方案1，记得在连接关闭或者组件销毁时，调用outgoingMessages$.complete()和socket$.unsubscribe()，避免内存泄漏哦～

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