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

如果 observable 的发射速度比循环完成的速度快，那么随着缓冲区越来越满，内存就会增加。我们可以提供使用不同策略的其他方法（例如，仅使用最近的值等），但将其保留为默认值。请注意，循环本身可能有多个等待，这会加剧问题。

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

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

Answer 1

可观察到的东西令人费解，我的理解可能存在缺陷。但是异步迭代器只是一个返回承诺的迭代器，它可以在“实时”事件流（热可观察）中解析未来的事件。它可以使用队列来实现，如下所示。

function* iterateClickEvents(target) {
  const queue = []
  target.addEventListener('click', e => queue.shift()?.fulfill(e))
  while (true)
    yield new Promise(fulfill => queue.push({fulfill}))
}

//use it
for await (const e of iterateClickEvents(myButton))
  handleEvent(e)

然后您可以实现流畅的运算符，例如：

class FluentIterable {
  constructor(iterable) {
    this.iterable = iterable
  }
  filter(predicate) {
    return new FluentIterable(this.$filter(predicate))
  }
  async* $filter(predicate) {
    for await (const value of this.iterable)
      if (predicate(value))
        yield value
  }
  async each(fn) {
    for await (const value of this.iterable)
      fn(value)
  }
}

//use it
new FluentIterable(iterateClickEvents(document.body))
  .filter(e => e.target == myButton)
  .each(handleEvent)
  .catch(console.error)

https://codepen.io/ken107/pen/PojZjgB

您可以实现一个map 运算符，它返回inner 迭代器的结果。事情从那里开始变得复杂。

Answer 2

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

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

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();
});

Answer 3

This 是当前的实现Observable[Symbol.asyncIterator]。

这是在数组上实现的Symbol.asyncIterator 的基本示例：

const dummyPromise = (val, time) => new Promise(res => setTimeout(res, time * 1000, val));

const items = [1, 2, 3];

items[Symbol.asyncIterator] = async function * () {
  yield * await this.map(v => dummyPromise(v, v));
}

!(async () => {
  for await (const value of items) {

  console.log(value);
}
})();
/* 
1 - after 1s
2 - after 2s
3 - after 3s
*/

我对生成器（sync 生成器）的理解是它们是可暂停函数，这意味着您可以立即请求一个值，然后在 10 秒后请求另一个值。异步生成器遵循相同的方法，除了它们生成的值是异步，这意味着您必须await 才能获得它。

例如：

const dummyPromise = (val, time) => new Promise(res => setTimeout(res, time * 1000, val));

const items = [1, 2, 3];
items[Symbol.asyncIterator] = async function * () {
  yield * await this.map(v => dummyPromise(v, v));
}

const it = items[Symbol.asyncIterator]();

(async () => {
  // console.log(await it.next())
  await it.next();

  setTimeout(async () => {
    console.log(await it.next());
  }, 2000); // It will take 4s in total
})();

---

回到Observable的实现：

async function* coroutine<T>(source: Observable<T>) {
  const deferreds: Deferred<IteratorResult<T>>[] = [];
  const values: T[] = [];
  let hasError = false;
  let error: any = null;
  let completed = false;

  const subs = source.subscribe({
    next: value => {
      if (deferreds.length > 0) {
        deferreds.shift()!.resolve({ value, done: false });
      } else {
        values.push(value);
      }
    },
    error: err => { /* ... */ },
    complete: () => { /* ... */ },
  });

  try {
    while (true) {
      if (values.length > 0) {
        yield values.shift();
      } else if (completed) {
        return;
      } else if (hasError) {
        throw error;
      } else {
        const d = new Deferred<IteratorResult<T>>();
        deferreds.push(d);
        const result = await d.promise;
        if (result.done) {
          return;
        } else {
          yield result.value;
        }
      }
    }
  } catch (err) {
    throw err;
  } finally {
    subs.unsubscribe();
  }
}

据我了解：

• values 用于跟踪同步值 如果您有of(1, 2, 3)，则values 数组将在到达while(true) { } 之前包含[1, 2, 3]。而且因为您使用的是for await (const v of ...)， 你会像在 it.next(); it.next(); it.next() ... 一样请求值。

  换句话说，一旦您可以从迭代器中使用一个值，您就会立即请求下一个值，直到数据生产者无法提供任何东西。

• deferreds 用于异步值 所以在你的第一个 it.next() ，values 数组是空的（意味着 observable 没有同步发出），所以它会回退到最后一个 else，它只是创建一个添加到 deferreds 的承诺, 之后，该承诺为 awaited，直到 resolves 或 rejects。

  当 observable 最终发出时，deferreds 不会为空，因此等待的 promise 将 resolve 与新到达的值。

const src$ = merge(
  timer(1000).pipe(mapTo(1)),
  timer(2000).pipe(mapTo(2)),
  timer(3000).pipe(mapTo(3)),
);

!(async () => {
  for await (const value of src$) {
    console.log(value);
  }
})();

StackBlitz