RxJS Subject多订阅者消息顺序异常问题及解决需求

问题场景 #

存在一个RxJS Subject的3个订阅者，其中第二个订阅者B在收到Subject的消息时，会向该Subject发送新消息。这导致第三个订阅者C接收消息的顺序错误——先收到B发送的消息，后收到初始消息。

原代码 #

import { Observable, Subject } from 'rxjs';

var mystream = new Subject<any>();

// first observer 
mystream.asObservable().subscribe(msg => console.log('A: ', msg));

// second observer which also adds a message to the subject
mystream.asObservable().subscribe(msg => {
  console.log('B: ', msg);

  if(msg === 'Initial message') {
    mystream.next('Message from B');
  }
});

// third observer who receives the messages as the last one and in incorrect order
mystream.asObservable().subscribe(msg => console.log('C: ', msg));

// start the whole process
mystream.next('Initial message');

当前输出 #

A:  Initial message
B:  Initial message
A:  Message from B
B:  Message from B
C:  Message from B
C:  Initial message

期望输出 #

A:  Initial message
B:  Initial message
C:  Initial message
A:  Message from B
B:  Message from B
C:  Message from B

问题原因 #

RxJS Subject的next()方法是同步执行的。当订阅者B在处理Initial message时调用mystream.next('Message from B')，会立刻触发所有订阅者处理这条新消息。此时订阅者C还未完成Initial message的处理流程，因此会先处理新消息，之后才回到Initial message的处理，导致顺序错乱。

解决方案 #

通过RxJS的queueScheduler将新消息的发送任务延迟到当前消息的所有订阅者处理完成后执行，确保消息顺序符合预期。

修改后的代码 #

import { Observable, Subject, queueScheduler } from 'rxjs';

var mystream = new Subject<any>();

// first observer 
mystream.asObservable().subscribe(msg => console.log('A: ', msg));

// second observer which also adds a message to the subject
mystream.asObservable().subscribe(msg => {
  console.log('B: ', msg);

  if(msg === 'Initial message') {
    // 利用queueScheduler调度新消息发送，确保当前消息的所有订阅者处理完毕后再执行
    queueScheduler.schedule(() => {
      mystream.next('Message from B');
    });
  }
});

// third observer
mystream.asObservable().subscribe(msg => console.log('C: ', msg));

// start the whole process
mystream.next('Initial message');

原理说明 #

queueScheduler会将任务加入到内部队列中，等当前同步执行的代码（即Initial message的所有订阅者处理逻辑）全部完成后，再依次执行队列中的任务。这样就能保证所有订阅者先处理完初始消息，再处理B发送的新消息，实现期望的顺序。

也可以使用asyncScheduler替代queueScheduler，两者的区别在于：

• queueScheduler属于微任务调度，执行时机更早（当前宏任务结束前）
• asyncScheduler属于宏任务调度，类似setTimeout，执行时机稍晚

两者都能解决顺序问题，可根据实际场景选择。

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