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

Question

我有一个Action,它接受一个回调,一旦它完成就会调用泛型参数,即一个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("\r\nCalling 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);

如果你这样做,你应该能够控制你的回调运行的线程.

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