RxJS内部机制疑问：Observable与Observer工作原理困惑

嘿，我来帮你理清这些RxJS的疑惑——其实核心是RxJS在背后悄悄帮我们做了不少包装工作，咱们逐个问题来拆解：

疑问1：Observer是在哪里、如何创建的？ #

当你调用observable.subscribe()时，RxJS会自动帮你创建Observer对象！

subscribe方法支持几种传入参数的方式：

• 传入一个完整的Observer对象（包含next/error/complete三个方法）
• 分别传入三个函数：subscribe(nextFn, errorFn, completeFn)
• 只传入一个函数（就像你示例里的x=>console.log(x)）

当你只传单个函数时，RxJS内部会把它包装成一个标准的Observer对象，大概长这样：

const observer = {
  next: x => console.log(x), // 你的回调函数
  error: () => {}, // 默认空实现
  complete: () => {} // 默认空实现
};

这个包装过程是RxJS内部自动完成的，你不用手动创建。

疑问2：是谁调用next方法向流中推送值？ #

分两种场景来看：

• 对于Rx.Observable.of('foo', 'bar')这种预定义值的Observable：它是冷Observable，只有当你调用subscribe时，内部才会开始执行逻辑——遍历你传入的'foo'和'bar'，依次调用Observer的next方法，最后触发complete。这个调用动作是of操作符的内部逻辑完成的。
• 对于Observable.create创建的Observable：你在create的回调函数里调用的observer.next()，其实是在调用RxJS帮你创建的那个Observer的next方法。而这个回调函数本身，是在你调用subscribe时被RxJS触发执行的。

疑问3：为什么x=>console.log(x)没有next方法却能正常运行？ #

这是RxJS提供的语法糖！

subscribe方法的设计就是为了简化使用：它不强制要求你传入完整的Observer对象。如果你只传一个函数，RxJS会默认把它当作Observer的next方法，自动补全error和complete的空实现，最终生成一个合法的Observer。

你也可以试试传多个参数，比如：

observable.subscribe(
  x => console.log(x), // next回调
  err => console.error(err), // error回调
  () => console.log('流结束了') // complete回调
);

这同样能正常运行，因为RxJS会把这三个函数包装成一个完整的Observer。

疑问4：Observable.create里的observer是哪来的？ #

当你调用myObservable.subscribe(value => console.log(value))时，RxJS会先完成前面说的“包装Observer”的步骤，生成一个带有next方法的Observer对象。

然后，RxJS会把这个Observer对象作为参数，传入你在Observable.create里定义的那个回调函数（也就是observer => { ... }里的observer参数）。所以你在里面调用的observer.next('foo')，本质上就是调用这个包装后的Observer的next方法，而这个next方法最终会触发你订阅时传入的value => console.log(value)回调。

我们可以用一段简单的代码模拟这个过程：

// 模拟RxJS的subscribe逻辑
function mockSubscribe(nextFn) {
  // 包装出完整的Observer
  const observer = {
    next: nextFn,
    error: () => {},
    complete: () => {}
  };
  // 执行create里的回调，把observer传进去
  mockCreateCallback(observer);
}

// 模拟你在create里写的回调
function mockCreateCallback(observer) {
  observer.next('foo');
  setTimeout(() => observer.next('bar'), 1000);
}

// 调用订阅，效果和RxJS示例一致
mockSubscribe(value => console.log(value));

这样是不是就清晰多了？RxJS的核心就是通过这些封装，让我们不用关心底层的Observer创建和调用细节，专注于数据流的处理~

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