反应式扩展中的嵌套订阅答案

【问题标题】：Nested Subscriptions in Reactive Extentions反应式扩展中的嵌套订阅
【发布时间】：2013-03-05 16:53:49
【问题描述】：

我无法让我的观察者在特定线程上正常工作。

Subject<bool> subjEvent = new Subject<bool>();
Subject<int> subjValue = new Subject<int>();

IScheduler sched = new EventLoopScheduler(ts => new Thread(ts));

subjEvent.ObserveOn(sched).Subscribe(
    r =>
        {
            if(r)
            {
                Console.WriteLine("Connected On: \t{0}", Thread.CurrentThread.ManagedThreadId);

                subjValue.ObserveOn(sched).Subscribe(
                    x => Console.WriteLine("Recieved On: \t{0}", Thread.CurrentThread.ManagedThreadId));
            }else{
                Console.WriteLine("Disconnect On: \t{0}", Thread.CurrentThread.ManagedThreadId);
            }
        }
    );

subjEvent.OnNext(true);
for(int i=0; i< 10; i++)
{
    subjValue.OnNext(i);
}
subjEvent.OnNext(false);
subjValue.OnCompleted();
subjEvent.OnCompleted();

这个想法是在某些东西可用时订阅它，然后取消订阅/重新订阅事件。现在，由于我需要在特定（读取相同）线程上进行观察，并确保正确排序，我有一个 EventLoopScheduler。

问题是我从价值订阅中一无所获。

现在，如果我将 Thread.Sleep(10) 添加到生成循环中（在 OnNext 之后），它可以正常工作。所以我对自己做错了什么感到困惑，非常感谢您的帮助/建议。

【问题讨论】：

标签： c# system.reactive

【解决方案1】：

虽然这无助于直接回答您的问题 - 我认为 Asti 在这方面做得很好 - 我想我会发布这段代码来帮助您编写查询。

您应该非常努力地不订阅您的 observables，除非您绝对必须这样做。当您在 OnNext 处理程序中执行代码时，需要尽可能小。

试试这个版本的查询：

var query =
    from b in subjEvent.Do(x =>
        Console.WriteLine("{0}onnected On: \t{1}",
            x ? "C" : "Disc",
            Thread.CurrentThread.ManagedThreadId))
    select b ? subjValue : Observable.Empty<int>();

query.Switch().Subscribe(x =>
    Console.WriteLine("Recieved On: \t{0}",
        Thread.CurrentThread.ManagedThreadId));

除非事件循环调度程序出现问题，否则此代码的功能与您的相同，但它只有一个订阅，并且在该订阅中做了最少的工作。我希望这会有所帮助。

【讨论】：

顺便说一下switch是干什么的？
@DavidHughes - query 返回一个 IObservable<IObservable<int>> 和 .Switch() 通过返回由外部 observable 的最新内部 observable 返回的一系列值来展平嵌套 observable - 它改变了类型IObservable<int> 这样做。

【解决方案2】：

这虽然是不确定的、预期的行为。问题是在EventLoopScheduler 中设置了第二个订阅。设置要被调度程序观察的订阅需要一些时间。但是它的通知已经在主线程中进行了。到订阅设置完成时，循环将完成执行。

要看到这种情况发生，请从您的活动主题中删除 ObserveOn。这将强制在继续向第二个主题通知值之前设置订阅。您也可以在OnNext(true) 之后等待一段时间，以便它连接订阅。

【讨论】：