为了实现您期望的行为，您需要为每个连接创建一个单独的流，并按您的意愿推送数据流。

下面是一个可能的极简解决方案

import { Controller, Get, MessageEvent, OnModuleDestroy, OnModuleInit, Res, Sse } from '@nestjs/common';
import { readFileSync } from 'fs';
import { join } from 'path';
import { Observable, ReplaySubject } from 'rxjs';
import { map } from 'rxjs/operators';
import { Response } from 'express';
@Controller()
export class AppController implements OnModuleInit, OnModuleDestroy {
  private stream: {
    id: string;
    subject: ReplaySubject<unknown>;
    observer: Observable<unknown>;
  }[] = [];
  private timer: NodeJS.Timeout;
  private id = 0;
  public onModuleInit(): void {
    this.timer = setInterval(() => {
      this.id += 1;
      this.stream.forEach(({ subject }) => subject.next(this.id));
    }, 1000);
  public onModuleDestroy(): void {
    clearInterval(this.timer);
  @Get()
  public index(): string {
    return readFileSync(join(__dirname, 'index.html'), 'utf-8').toString();
  @Sse('sse')
  public sse(@Res() response: Response): Observable<MessageEvent> {
    const id = AppController.genStreamId();
    // Clean up the stream when the client disconnects
    response.on('close', () => this.removeStream(id));
    // Create a new stream
    const subject = new ReplaySubject();
    const observer = subject.asObservable();
    this.addStream(subject, observer, id);
    return observer.pipe(map((data) => ({
      id: `my-stream-id:${id}`,
      data: `Hello world ${data}`,
      event: 'my-event-name',
    }) as MessageEvent));
  private addStream(subject: ReplaySubject<unknown>, observer: Observable<unknown>, id: string): void {
    this.stream.push({
      subject,
      observer,
  private removeStream(id: string): void {
    this.stream = this.stream.filter(stream => stream.id !== id);
  private static genStreamId(): string {
    return Math.random().toString(36).substring(2, 15);
}

您可以为它创建一个单独的服务，并使其更干净，并从不同的地方推送流数据，但作为一个示例，这将导致如下屏幕截图所示的结果。

​

​

这种行为是正确的。每个SSE连接都是一个专用套接字，由专用服务器进程处理。这样每个客户端就可以接收到不同的数据。这不是一种广播一样的技术。

我怎样才能得到预期的行为呢？

将您想要发送到所有连接客户端的所需值的中央记录(例如，在SQL中)。

然后让每个SSE服务器进程监视或轮询中央记录，并在每次更改时发送一个事件。

您只需为同一主题的每个sse连接生成一个新的可观察的

 private events: Subject<MessageEvent> = new Subject();
 constuctor(){
   timer(0, 1000).pipe(takeUntil(this.destroy)).subscribe(async (index: any)=>{
            let event: MessageEvent = { 
                id: index, 
                type: 'test', 
                retry: 30000, 
                data: {index: index}
            } as MessageEvent;
            this.events.next(event);