javascript - 如何使 RxJS Observable 在特定日期时间发出？

Question

我有一个 RxJS Observable需要在特定时间重新计算，如 DateTime 数组所述对象(尽管出于这个问题的目的，它们可以是 JavaScript Date 对象、纪元毫秒或任何其他代表特定时刻的东西):

const changeTimes = [
    //            yyyy, mm, dd, hh, mm
    DateTime.utc( 2018, 10, 31, 21, 45 ),
    DateTime.utc( 2018, 10, 31, 21, 50 ),
    DateTime.utc( 2018, 10, 31, 22, 00 ),
    DateTime.utc( 2018, 10, 31, 23, 00 ),
    DateTime.utc( 2018, 10, 31, 23, 30 ),
];

我正在努力理解如何创建一个在这样一个数组中指定的时间发出的 Observable。

这是我在尝试回答我自己的问题时的想法:

我几乎肯定需要使用 the delay operator其中指定的延迟是“现在”和下一个 future 日期时间之间的时间。

我需要确保“现在”在订阅时是最新的，而不是在创建 Observable 时——可能通过使用 the defer operator ——尽管如果有多个订阅，我不想不必要地创建多个 Observable 实例。

随着时间的推移，我不确定如何迭代数组— the expand operator可能是我需要的，但它递归地调用了一些东西，我只是想遍历一个列表。

The timer operator似乎无关紧要，因为每个日期时间之间的持续时间不同。

我可以将每个日期时间映射到它自己的延迟 Observable 并通过 merge 将它们全部返回。 ，但是随着数组中日期时间数量的增加(可能有数百个)，这变得非常低效，所以这绝对是最后的手段。

如何制作一个 RxJS Observable，它需要一个日期时间列表，然后在每个时间到达时发出，最后一个完成？

Answer 1

我认为您在要点中总结的内容都是正确的。使用 delay看起来很明显，但它会使链条难以理解。

我想到的解决方案假设您知道 changeTimes创建可观察链之前的数组。您可以创建自己的“可观察创建方法”，该方法将返回基于 setTimeout 发出的可观察对象例如(这只是“伪代码”，它不能正确计算日期):

const schedule = (dates: Date[]): Observable<Date> => new Observable(observer => {
  // Sort the `dates` array from the earliest to the latest...

  let index = 0;
  let clearTimeout;

  const loop = () => {
    const now = new Date();
    const delay = dates[index] - now;

    clearTimeout = setTimeout(() => {
      observer.next(dates[index++]);

      if (index < dates.length) {
        loop();
      }
    }, delay);
  }

  loop();

  return () => clearTimeout(clearTimeout);
}); 

...

schedule(changeTimes)
  .subscribe(...)

您提到的最后一个选项 merge实际上不是那么糟糕。我了解您担心它会创建大量订阅，但如果您对 changeTimes 进行排序数组，然后使用 concat而不是 merge即使您创建了 100 个 Observable，它也将始终只保留一个事件订阅。