javascript - AsyncIterable 和 Observable 之间的实际区别是什么？

Question

我最近一直被这个话题挂断。似乎 AsyncIterables 和 Observables 都具有类似流的特性，尽管它们的使用方式略有不同。

您可以像这样使用异步迭代

const myAsyncIterable = async function*() { yield 1; yield 2; yield 3; }

const main = async () => {
  for await (const number of myAsyncIterable()) {
    console.log(number)
  }
}

main()

您可以像这样使用 observable

const Observable = rxjs
const { map } = rxjs.operators

Observable.of(1, 2, 3).subscribe(x => console.log(x))

<script src="https://unpkg.com/rxjs/bundles/rxjs.umd.min.js"></script>

我的首要问题是基于这个 RxJS pr

If the observable emits at a pace faster than the loop completes, there will be a memory build up as the buffer gets more full. We could provide other methods that use different strategies (e.g. just the most recent value, etc), but leave this as the default. Note that the loop itself may have several awaits in it, that exacerbate the problem.

在我看来，异步迭代器本质上没有背压问题，所以实现 Symbol.asyncIterator 是否正确(@@asyncIterator) 在 Observable 上并默认为背压策略？鉴于 AsyncIterables，甚至还需要 Observables 吗？

理想情况下，您可以通过代码示例向我展示 AsyncIterables 和 Observables 之间的实际差异。

Answer 1

主要区别在于哪一方决定何时迭代。

在异步迭代器的情况下，客户端通过调用 await iterator.next() 来决定。 .源决定何时解决 promise ，但客户端必须首先要求下一个值。因此，消费者从源中“拉取”数据。

Observables 注册一个回调函数，当有新值进入时，observable 立即调用该回调函数。因此，源“推送”给消费者。

通过使用 Subject 并将其映射到异步迭代器的下一个值，可以轻松地使用 Observable 来使用异步迭代器。然后，只要您准备好使用下一个项目，您就可以在主题上调用 next。这是一个代码示例

const pull = new Subject();
const output = pull.pipe(
  concatMap(() => from(iter.next())),
  map(val => { 
    if(val.done) pull.complete();
    return val.value;
  })
);
//wherever you need this 
output.pipe(

).subscribe(() => {
  //we're ready for the next item
  if(!pull.closed) pull.next();
});