观察回到当前线程

Question

我正在尝试在新线程上安排一个可观察对象并在当前线程上取回结果。

namespace ConsoleApplication1
{
    class Program
    {
        static void Main(string[] args)
        {
            Log("init");

            Observable.Return(0)
                .ObserveOn(NewThreadScheduler.Default)
                .Do(_ => Log("Do1 method"))
                .ObserveOn(CurrentThreadScheduler.Instance)
                .Do(_ => Log("Do2 method"))
                .Subscribe(_ => Log("subscribe method"));

            Console.ReadKey();
        }

        static void Log(string label)
        {
            Console.WriteLine("{0} on {1}", label, Thread.CurrentThread.ManagedThreadId);
        }
    }
}

这是我得到的结果：

init on 9
Do1 method on 10
Do2 method on 10
subscribe method on 10

为什么 Do2 方法和订阅方法在线程 #10 而不是 #9 上？我期待这个结果：

init on 9
Do1 method on 10
Do2 method on 9
subscribe method on 9

Answer 1

我明白您要实现的目标，但这可能会证明很困难，因为您只是运行在同步线程上的控制台应用程序中。由于主控制台线程不是 EventLoop，Rx 如何才能 "unblock" Console.ReadKey() 调用，劫持线程执行 Log 方法然后 return 等待Console.ReadKey()?

如果您在 GUI 应用程序中执行此操作，那么主线程将是某种事件循环，即 WPF/Silverlight 中的 Dispatcher。这现在很容易 return 控制回到 "main thread" - ObserveOn(_dispatcherScheduler) 其中 _dispatcherScheduler 是 DispatcherScheduler.

的捕获实例

您可以在此处查看 Immediate 和 Current 调度器的解释 - http://introtorx.com/Content/v1.0.10621.0/15_SchedulingAndThreading.html#SchedulersIndepth.

原始代码的替代方法是运行第三个线程使用完美命名的 EventLoopScheduler

作为 EventLoop

void Main()
{
    var els = new EventLoopScheduler(ts => new Thread(ts) { IsBackground = true, Name = "MyEventLoopThread"});
    els.Schedule(0, (scheduler, _)=>Run(scheduler));
}
static IDisposable Run(IScheduler mainThreadScheduler)
{
    Log("init");
    return Observable.Return(0)
        .ObserveOn(NewThreadScheduler.Default)
        .Do(_ => Log("Do1 method"))
        .ObserveOn(mainThreadScheduler)
        .Do(_ => Log("Do2 method"))
        .Subscribe(_ => Log("subscribe method"));
}
static void Log(string label)
{
    Console.WriteLine("{0} on {1}", label, Thread.CurrentThread.ManagedThreadId);
}

创建以下（预期）输出

init on 28
Do1 method on 29
Do2 method on 28
subscribe method on 28

观察回到当前线程

Observe back to current thread

c#

concurrency

system.reactive