Angular 5中Rxjs Subject.subscribe()在多组件中仅单组件触发问题

看起来你遇到的问题是HeaderComponent无法接收到NotificationService推送的通知，而NotificationsComponent可以正常响应。我帮你分析一下核心问题和解决办法：

核心问题1：错误地将Observable强制转换为ReplaySubject #

在你的NotificationService中，你把WebSocket返回的ReplaySubject<MessageEvent>经过map转换后，强制类型转换成了ReplaySubject<Notification>：

this._notification = <ReplaySubject<Notification>>websocketService
 .connect(NOTIFICATION_URL)
 .map(...);

但map操作符返回的是普通Observable，不是Subject的子类，它没有next()方法。你这里的强制类型转换是不安全的，虽然NotificationsComponent能收到WebSocket的消息（原Subject的事件会流到这个Observable），但当你调用this._notification.next(notification)时，实际上是在调用一个不存在的方法——这也是HeaderComponent订阅异常的潜在原因之一。

核心问题2：服务可能不是单例 #

如果你的NotificationService或WebSocketService在多个模块/组件的providers数组中重复声明了，Angular会为每个声明的模块创建独立的服务实例。这意味着NotificationsComponent和HeaderComponent注入的是两个不同的服务实例，自然无法共享通知流。

分步解决方案 #

1. 修复WebSocketService的实现 #

先修正WebSocketService中ReplaySubject的创建方式，让逻辑更清晰：

import { Injectable } from '@angular/core';
import { ReplaySubject, Observable, Observer } from 'rxjs';

@Injectable({ providedIn: 'root' }) // 标记为根注入，确保全局单例
export class WebsocketService {
  private subject: ReplaySubject<MessageEvent> | null = null;

  constructor() { }

  public connect(url: string): ReplaySubject<MessageEvent> {
    if (!this.subject) {
      this.subject = this.createConnection(url);
      console.log("Successfully connected: " + url);
    }
    return this.subject;
  }

  private createConnection(url: string): ReplaySubject<MessageEvent> {
    const ws = new WebSocket(url);
    const subject = new ReplaySubject<MessageEvent>();

    // 转发WebSocket事件到Subject
    ws.onmessage = (event) => subject.next(event);
    ws.onerror = (error) => subject.error(error);
    ws.onclose = () => subject.complete();

    // 处理发送逻辑：订阅Subject的next来发送消息
    subject.subscribe({
      next: (data: any) => {
        if (ws.readyState === WebSocket.OPEN) {
          console.log("---sending ws message---");
          ws.send(JSON.stringify(data));
        }
      }
    });

    return subject;
  }
}

2. 重构NotificationService，拆分发送和接收流 #

现在把发送和接收的流分开，避免类型转换错误：

import { Injectable } from '@angular/core';
import { Observable, ReplaySubject } from 'rxjs';
import { map } from 'rxjs/operators';
import { WebsocketService } from './websocket.service';
import { Notification } from './../model/notification'

const NOTIFICATION_URL = 'ws://localhost:8080/Kwetter/socket';

@Injectable({ providedIn: 'root' }) // 根注入确保单例
export class NotificationService {
  private readonly wsSubject: ReplaySubject<MessageEvent>;
  public readonly notifications$: Observable<Notification>; // 用于组件订阅接收通知

  constructor(websocketService: WebsocketService) {
    this.wsSubject = websocketService.connect(NOTIFICATION_URL);
    // 转换消息格式，提供给组件订阅
    this.notifications$ = this.wsSubject.pipe(
      map((response: MessageEvent): Notification => {
        const data = JSON.parse(response.data);
        return {
          sender: data.author,
          message: data.message
        };
      })
    );
  }

  sendMessage(notification: Notification) {
    console.log("---calling .next()---");
    this.wsSubject.next(notification); // 用原WebSocket Subject发送消息
  }
}

3. 修正组件中的订阅代码 #

更新两个组件的订阅逻辑，使用新的notifications$ observable：

NotificationsComponent

constructor(private notificationService: NotificationService, private userService: UserService) {
  if (this.notification == null) {
    this.notification = new Notification("", "");
  }
  // 订阅新的notifications$流
  notificationService.notifications$.subscribe(notification => {
    console.log("---notification has been updated---")
    this.notification = notification;
  });
}

HeaderComponent

constructor(private userService: UserService, private router: Router, private notificationService: NotificationService) {
  console.log("---constructor headercomponent---");
  console.log(this.notification);
  // 订阅新的notifications$流
  this.subscription = this.notificationService.notifications$.subscribe(notification => {
    console.log("---header notification updated---");
    this.notification = notification;
  });
}

4. 确保服务是单例 #

移除所有组件/子模块中对NotificationService和WebSocketService的providers声明，只保留在根模块（AppModule）或者使用providedIn: 'root'（上面的代码已经配置），这样整个应用只会有一个服务实例。

验证 #

现在重新运行应用：

1. 在NotificationsComponent发送消息，控制台应该能看到---header notification updated---日志
2. 其他浏览器的实例也能同步收到消息
3. 不会再出现类型错误或无效的next()调用

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