c# - 与多个 IEnumerables 一起使用时 Observable.Subscribe 的行为

Question

我正在尝试创建一个 IObservable<T>来自两个数组 ( IEnumerable s)。我试图避免显式迭代数组并调用 observer.OnNext .我遇到了 Observable.Subscribe扩展方法，乍一看似乎是我需要的。但是，它没有像我预期的那样工作，我不知道为什么。

下面的代码是一个例子:

  class Program
  {
    static void Main(string[] args)
    {
      var observable = Observable.Create<char>(observer =>
        {
          var firstBytes = new[] {'A'};
          var secondBytes = new[] {'Z', 'Y'};
          firstBytes.Subscribe(observer);
          secondBytes.Subscribe(observer);

          return Disposable.Empty;
        }
      );

      observable.Subscribe(b => Console.Write(b));
    }
  }

它的输出是“AZ”，而不是我预期的“AZY”。现在，如果我订阅 secondBytes之前 firstBytes ，输出为“ZAY”!这似乎表明它正在同步枚举两个数组——这在某种程度上解释了“AZ”输出。

无论如何，我完全不知道为什么它会这样，并且希望人们能够提供任何见解。

Answer 1

锁步迭代行为的原因可以通过 Observable.Subscribe(IEnumerable source) 的实现来解释。它使用 "recursive" algorithm它通过在调度程序操作中调用 e.MoveNext 来工作。如果成功，则发出该值，然后新的调度程序操作排队以从可枚举中读取下一个值。

由于您订阅了两个可枚举对象并且没有为订阅指定任何特定的调度程序，因此默认迭代调度程序将用于这些操作(由 SchedulerDefaults.Iteration 定义)，默认运行于当前线程。这意味着枚举操作将排队等候在您当前的订阅操作完成后运行。这导致枚举操作交错 - 像这样

1. firstBytes.Subscribe() -> 队列枚举操作
2. secondBytes.Subscribe() -> 队列枚举操作
3. 调用 firstBytes.MoveNext() -> OnNext("A") -> 排队下一个枚举操作
4. 调用 secondBytes.MoveNext() -> OnNext("Z") -> 排队下一个枚举操作
5. 调用 firstBytes.MoveNext() -> OnCompleted()
6. 调用 secondBytes.MoveNext() -> OnNext(Y) -> 排队下一个枚举操作
7. 调用 secondBytes.MoveNext() -> OnCompleted()

观察者在第 5 步收到 OnCompleted() 通知，因此忽略剩余的 secondBytes 枚举步骤。如果您退回了订阅的一次性用品，那么此时将取消第二次枚举。