ObserveOn(Scheduler.CurrentThread) 不会导致订阅的操作在原始线程上运行

Question

我有一个动作，它接受一个回调，一旦完成，它将用一个通用参数调用，即一个Action<Action<T>>。我想在操作开始时显示一个繁忙的微调器，然后在它调用回调时将其拿走，所以我创建了一个简单的实用程序来执行此操作。我遇到的问题是用户希望他们的回调在原始调用线程上运行，但它并不总是这样做。它几乎总是在单元测试（nUnit）中完美运行，但在应用程序实际运行时（WPF，.Net 4）不适用于某些调用。

以下是我所拥有的相关内容

void WrapAsyncCallbackPattern<T>(Action<T> callback, Action<Action<T>> actionToRun)
{
    var subject = new AsyncSubject<T>();
    try
    {
        actionToRun(
            result =>
            {
                subject.OnNext(result);
                subject.OnCompleted();
            });
    }
    catch (Exception ex)
    {
        subject.OnError(ex);
    }
    subject
        .ObserveOn(Scheduler.CurrentThread)
        .Subscribe(callback, OnError);
}

我希望callback在我订阅的线程上运行（并且声明了subject），但它似乎并不可靠地运行。我假设我在做一些愚蠢的事情。这是什么？

编辑：添加了单元测试代码

private readonly TimeSpan m_WaitTime = TimeSpan.FromSeconds(1);
[Test]
public void WrapAsyncCallbackPattern_WithActionOnDifferentThread_CallsCallbackOnSameThread()
{
    var awaiter = new AutoResetEvent(false);
    bool callbackRan = false;
    int callingThreadId = Thread.CurrentThread.ManagedThreadId;
    int callbackThreadId = int.MinValue;
    int actionThreadId = int.MinValue;
    BackgroundOperation.WrapAsyncCallbackPattern<int>(
        _ =>
        {
            callbackRan = true;
            callbackThreadId = Thread.CurrentThread.ManagedThreadId;
            awaiter.Set();
        },
        cb => ThreadPool.QueueUserWorkItem(
            _ =>
            {
                actionThreadId = Thread.CurrentThread.ManagedThreadId;
                cb(0);
            }));
    var errorInfo = string.Format("rnCalling thread = {0}; Action thread = {1}; Callback thread = {2}", callingThreadId, actionThreadId, callbackThreadId);
    Assert.IsTrue(awaiter.WaitOne(m_WaitTime));
    Assert.IsTrue(callbackRan);
    Assert.AreNotEqual(callingThreadId, actionThreadId, "Action needs to be run on a different thread for this test." + errorInfo);
    Assert.AreNotEqual(actionThreadId, callbackThreadId, "Callback should not be run on action thread." + errorInfo);
    Assert.AreEqual(callingThreadId, callbackThreadId, "Callback should be run on calling thread." + errorInfo);
}

Answer 1

你可能对Scheduler.CurrentThread做什么有错误的理解。我认为每个人都会犯这个错误。

CurrentThread调度程序与执行的可观察量相关，而不是何时定义（或订阅）。考虑延迟或懒惰的执行。这应该是有意义的，因为每当您跳转到不同的线程时，您都需要某种方式来编组调用。

所以你真正追求的是这样的东西：

var synchContext = new SynchronizationContextScheduler(
    System.Threading.SynchronizationContext.Current) 
subject
    .ObserveOn(synchContext)
    .Subscribe(callback, OnError);

或者也许：

subject
    .ObserveOn(this) /* this is my current form */
    .Subscribe(callback, OnError);

如果你这样做，你应该能够控制你的回调在哪个线程上运行。

您的测试可能有效，因为它们最终同步执行。