反应式扩展：为什么会立即退出？

Question

我正在阅读IntroToRx，但示例代码有点问题。这是我的代码的总和：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace LearningReactiveExtensions
{
  public class Program
  {
    static void Main(string[] args)
    {
        var observable = Observable.Interval(TimeSpan.FromSeconds(5));
        observable.Subscribe(
          Console.WriteLine, 
          () => Console.WriteLine("Completed")
        );
        Console.WriteLine("Done");
        Console.ReadKey();
    }

  }
}

如果我对这本书的理解正确，这应该会向控制台写入一个数字序列，每五秒一次永远，因为我从来没有Dispose()这个序列。

但是，当我运行代码时，我得到的只是最后的“完成”。没有数字，没有“完成”，只有“完成”。

我在这里做错了什么？

Answer 1

我假设您没有耐心等待 5 秒钟，否则您会看到代码正在运行。

要记住Rx 的主要思想是Observable.Subscribe 几乎会立即将控制权返回给调用方法。换句话说，Observable.Subscribe 在结果产生之前不会阻塞。因此对Console.WriteLine 的调用将仅在五秒后被调用。

Answer 2

你需要一些方法让主线程等待你正在做的事情。如果您愿意，可以使用信号量

using System;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace LearningReactiveExtensions
{
  public class Program
  {
    static void Main(string[] args)
    {
         SemaphoreSlim ss = new SemaphoreSlim(1);
        var observable = Observable.Interval(TimeSpan.FromSeconds(5));
        observable.Subscribe(
          Console.WriteLine, 
          () => {
               Console.WriteLine("Completed");
               ss.Release();
          }
        );
        ss.Wait();
        Console.WriteLine("Done");
        Console.ReadKey();
    }

  }
}

虽然在这种情况下写起来可能更好

  static void Main(string[] args)
   {
        SemaphoreSlim ss = new SemaphoreSlim(1);
        Observable.Interval(TimeSpan.FromSeconds(5)).Wait();
        Console.WriteLine("Completed");
        Console.WriteLine("Done");
        Console.ReadKey();
   }