Observable.Using 在其他线程上处理而不是在创建的线程上

Question

我有一个 ASP.NET WebApi 请求方法，它反过来在旧资源上使用 Observable.Using 启动异步调用。此资源将生成一个新线程，在该线程上引发事件，这些事件又转换为 IObservable 流中的 OnNext 项，该流由资源周围的包装器公开。

我是awaiting 使用IObservable.FirstOrDefaultAsync() 的流结果，我的WebApi 方法标记为async。

如果我采用这种设置，我会得到臭名昭著的

异步模块或处理程序 在异步操作仍处于挂起状态时完成

所以，我的第一个问题是关于这个的。我假设我得到了这个，因为新的异步操作（没有async/await）是由旧资源产生的，但是 ASP.NET 究竟是如何知道这一点的呢？注册了什么？我发现 ASP.NET 正在查看 SynchronizationContext.Current._state.VoidAsyncOutstandingOperationCount 以引发此异常，但哪些调用会增加此属性？排队到 ThreadPool 的线程？我相当肯定这是由遗留资源制作的。

现在为什么在我处置资源时这些操作仍在进行？好吧，看来Dispose 是在传播事件的线程上运行的，在概念上类似于下面的 sn-p。

Observable
    .Using(
        () => {
            Console.WriteLine($"Created on thread: {Thread.CurrentThread.ManagedThreadId}");
            return Disposable.Create(() => {
                Console.WriteLine($"Disposed on thread: {Thread.CurrentThread.ManagedThreadId}");
            });
        },
        _ => Observable.Return(1, NewThreadScheduler.Default))
    .Do(_ => Console.WriteLine($"OnNext on thread: {Thread.CurrentThread.ManagedThreadId}"))
    .Wait();

类似这样的结果：

Created on thread: 10
OnNext on thread: 11
Disposed on thread: 11

这是设计使然吗？当使用using 释放资源时，很明显该资源是在创建它的同一线程上释放的，因为代码是同步的。随着Dispose 在单独的线程上运行，调用代码将继续运行，并且控制器将在Dispose 完全完成之前返回（至少在大多数情况下）。

我怎样才能以理智的方式缓解这种情况？一种似乎可行的方法是，而不是使用Observable.Using，而是使用返回给控制器的这个构造，awaiting 它使用FirstOrDefaultAsync()：

var resource = // Creating resource manually.
return resource.StartAsyncOperation() // <- observable producing events
    .ObserveOn(SynchronizationContext.Current)
    .Do(_ => resource.Dispose());

不过，这对我来说就像是一种黑客行为。

想法和建议？

编辑 1

我想我在这里面临的问题之一是资源的Dispose 方法在我使用Observable.Using 时被调用之后序列被终止/完成。应该是这样吗？在这种情况下，真的没有办法等待Dispose 使用该构造。我必须使用额外的 IObservable<Unit> Disposed() 方法或类似的方法来修改 api...

Answer 1

Dispose 将在调用它的线程上被调用（嗯，呵呵）。为了更有帮助，当在另一个线程上使用/观察 Rx 序列时，OnComplete 回调将被调用。如果您将 Rx 与标准运算符一起使用，那么您将在序列终止时获得自动处理行为（使用 OnError 或 OnComplete）。这种自动处理将在OnComplete/OnError 之后直接发生，并且只会在同一个线程上运行。

如果您希望您的处置绑定到调度程序，那么我建议查看 System.Reactive.Disposables.ScheduledDisposable 类型。然而，在这里使用SynchronizationContext 似乎更自然，所以在这种情况下System.Reactive.Disposables.ContextDisposable 可能更合适。