使用 Rx 阻塞（并可能超时）异步操作

Question

我正在尝试使用 .NET 响应式扩展重写一些代码，但我需要一些关于如何实现我的目标的指导。

我有一个类将一些异步行为封装在低级库中。考虑一些可以读取或写入网络的东西。当类启动时，它将尝试连接到环境，成功后，它将通过从工作线程调用返回信号。

我想将此异步行为转换为同步调用，并且我在下面创建了一个大大简化的示例来说明如何实现这一点：

ManualResetEvent readyEvent = new ManualResetEvent(false);

public void Start(TimeSpan timeout) {
  // Simulate a background process
  ThreadPool.QueueUserWorkItem(_ => AsyncStart(TimeSpan.FromSeconds(1)));
  // Wait for startup to complete.
  if (!this.readyEvent.WaitOne(timeout))
    throw new TimeoutException();
}

void AsyncStart(TimeSpan delay) {
  Thread.Sleep(delay); // Simulate startup delay.
  this.readyEvent.Set();
}

在工作线程上运行 AsyncStart 只是模拟异步行为的一种方法库，不是我的真实代码的一部分，其中低级库提供线程并在回调中调用我的代码。

请注意，如果启动未在超时间隔内完成，Start 方法将抛出 TimeoutException。

我想重写这段代码以使用 Rx。这是我的第一次尝试：

Subject<Unit> readySubject = new Subject<Unit>();

public void Start(TimeSpan timeout) {
  ThreadPool.QueueUserWorkItem(_ => AsyncStart(TimeSpan.FromSeconds(1)));
  // Point A - see below
  this.readySubject.Timeout(timeout).First();
}

void AsyncStart(TimeSpan delay) {
  Thread.Sleep(delay);
  this.readySubject.OnNext(new Unit());
}

这是一次不错的尝试，但不幸的是它包含竞争条件。如果启动快速完成（例如，如果delay为0）并且如果在A点有额外的延迟，则OnNext将在< code>readySubject 在 First 执行之前。本质上，我正在应用 Timeout 和 First 的 IObservable 永远不会看到启动已完成，并且会出现 TimeoutException改为抛出。

似乎 Observable.Defer 就是为了处理这样的问题而创建的。下面是使用 Rx 的稍微复杂一点的尝试：

Subject<Unit> readySubject = new Subject<Unit>();

void Start(TimeSpan timeout) {
  var ready = Observable.Defer(() => {
    ThreadPool.QueueUserWorkItem(_ => AsyncStart(TimeSpan.FromSeconds(1)));
    // Point B - see below
    return this.readySubject.AsObservable();
  });
  ready.Timeout(timeout).First();
}

void AsyncStart(TimeSpan delay) {
  Thread.Sleep(delay);
  this.readySubject.OnNext(new Unit());
}

现在异步操作不会立即启动，而是仅在使用 IObservable 时启动。不幸的是，仍然存在竞争条件，但这次是在 B 点。如果异步操作在 Defer lambda 返回之前开始调用 OnNext，它仍然会丢失并出现 TimeoutException将被Timeout抛出。

我知道我可以使用像 Replay 这样的运算符来缓冲事件，但我最初的没有 Rx 的示例不使用任何类型的缓冲。有没有办法让我在没有竞争条件的情况下使用 Rx 来解决我的问题？本质上，只有在连接到IObservable（在这种情况下Timeout和First）之后才开始异步操作？

---

根据 Ana Betts 的回答，这里是可行的解决方案：

void Start(TimeSpan timeout) {
  var readySubject = new AsyncSubject<Unit>();
  ThreadPool.QueueUserWorkItem(_ => AsyncStart(readySubject, TimeSpan.FromSeconds(1)));
  // Point C - see below
  readySubject.Timeout(timeout).First();
}

void AsyncStart(ISubject<Unit> readySubject, TimeSpan delay) {
  Thread.Sleep(delay);
  readySubject.OnNext(new Unit());
  readySubject.OnCompleted();
}

有趣的部分是，C 点的延迟时间长于 AsyncStart 完成所需的时间。 AsyncSubject 保留最后发送的通知，并且 Timeout 和 First 仍将按预期执行。

Answer 1

因此，我认为很多人一开始（包括我自己！）都需要了解 Rx 的一件事：如果您使用任何传统的线程函数，如 ResetEvents、Thread.Sleeps 或其他任何函数，那么您就错了（tm ) - 这就像在 LINQ 中将事物转换为数组，因为您知道底层类型恰好是数组。

要知道的关键一点是，异步函数由返回 IObservable的函数表示 - 这是让您在某件事完成时发出信号的魔力。因此，以下是如何“Rx-ify”更传统的异步函数，就像您在 Silverlight Web 服务中看到的那样：

IObservable<byte[]> readFromNetwork()
{
    var ret = new AsyncSubject();
    // Here's a traditional async function that you provide a callback to
    asyncReaderFunc(theFile, buffer => {
        ret.OnNext(buffer);
        ret.OnCompleted();
    });

    return ret;
}

这是一次不错的尝试，但不幸的是它包含竞争条件。

这就是 AsyncSubject 发挥作用的地方 - 这确保即使 asyncReaderFunc 击败了 Subscribe，AsyncSubject 仍然会“重播”发生的事情。

所以，现在我们已经有了我们的函数，我们可以对其做很多有趣的事情：

// Make it into a sync function
byte[] results = readFromNetwork().First();

// Keep reading blocks one at a time until we run out
readFromNetwork().Repeat().TakeUntil(x => x == null || x.Length == 0).Subscribe(bytes => {
    Console.WriteLine("Read {0} bytes in chunk", bytes.Length);
})

// Read the entire stream and get notified when the whole deal is finished
readFromNetwork()
    .Repeat().TakeUntil(x => x == null || x.Length == 0)
    .Aggregate(new MemoryStream(), (ms, bytes) => ms.Write(bytes))
    .Subscribe(ms => {
        Console.WriteLine("Got {0} bytes in total", ms.ToArray().Length);
    });

// Or just get the entire thing as a MemoryStream and wait for it
var memoryStream = readFromNetwork()
    .Repeat().TakeUntil(x => x == null || x.Length == 0)
    .Aggregate(new MemoryStream(), (ms, bytes) => ms.Write(bytes))
    .First();

Answer 2

我想进一步补充保罗的评论，添加 WaitHandles 意味着你做错了，直接使用主题通常也意味着你做错了。 ;-)

尝试考虑您的 Rx 代码与序列或管道一起使用。主题提供读写功能，这意味着您不再使用管道或序列（除非您有双向管道或可以反转的序列？！？）

所以首先保罗的代码非常酷，但是让我们“Rx见鬼去吧”。

1st AsyncStart 方法将其更改为

IObservable<Unit> AsyncStart(TimeSpan delay) 
{
  Observable.Timer(delay).Select(_=>Unit.Default);
}

如此简单！看起来没有主题，数据仅以一种方式流动。这里重要的是签名的更改。它会把东西推给我们。现在这一点已经非常明确了。向我传递一个主题是非常含糊的。

第二。我们现在不需要在 start 方法中定义的主题。我们还可以利用 Scheduler 功能来代替旧的 skool ThreadPool.QueueUserWorkItem。

void Start(TimeSpan timeout) 
{
    var isReady = AsyncStart(TimeSpan.FromSeconds(1))
                    .SubscribeOn(Scheduler.ThreadPool)
                    .PublishLast();
    isReady.Connect();
    isReady.Timeout(timeout).First();
}

现在我们有了一个清晰的管道或事件序列

AsyncStart -->已准备好 --> Start

而不是 Start-->AsyncStart-->Start

如果我对您的问题空间有更多了解，我相信我们可以想出一种更好的方法来执行此操作，并且不需要 start 方法的阻塞性质。使用 Rx 越多，您就越会发现关于何时需要阻塞、使用等待句柄等的旧假设可以被抛到九霄云外。