Xamarin Android中的SubscribeOn / ObserveOn示例

Question

我尝试在android Xamarin中使用RX。
你知道如何使用:
Observable.SubscribeOn(..)声明使用默认池中的线程执行后台任务Observable.ObserveOn(..)将事件从Observable重定向到UI线程

根据SubscribeOn，我尝试了所有的调度程序从
System.Reactive.PlatformServices.dll / System.Reactive.Concurrency
即:

NewThreadScheduler.Default
TaskPoolScheduler.Default
ThreadPoolScheduler.Instance

却什么都没起作用

另一方面[如果可能]我不想手动:
在Observable中创建我的线程
-在Observer中使用runonuthread

= = = update = =

NewThreadScheduler.Default检测结果
代码:

Console.WriteLine("creating ole");
var ole = Observable.Create<string>(suber =>
{
   Console.WriteLine("inside ole");
   Thread.Sleep(5000);
   suber.OnNext("point1");
   suber.OnCompleted();
   Console.WriteLine("ole completed");
   return Disposable.Create(() => Console.WriteLine("observer unsubscribed"));
});
ole.SubscribeOn(NewThreadScheduler.Default);
Console.WriteLine("subscribing");
oleSub = ole.Subscribe(s => Console.WriteLine("result: " + s));
Console.WriteLine("subscribed");

输出:

creating ole
subscribing
inside ole
result: point1
ole completed
observer unsubscribed
subscribed

结论:
可观察对象的内容在主线程中执行，尽管期望运行它自己的线程:ole.SubscribeOn(NewThreadScheduler.Default);

Answer 1

Rx基于函数式编程，其中一个关键特点是无副作用编程。

当你创建var ole = Observable.Create...时，你正在创建一个可观察序列。当有内容订阅时，此序列将具有行为。

当您尝试在ole.SubscribeOn(NewThreadScheduler.Default);上设置订阅时，您正在用SubscribeOn行为装饰ole可观察序列，然而，由于这是一个无副作用的操作，它返回一个新的可观察序列。您不将返回的实例分配给任何东西。即ole.SubscribeOn(NewThreadScheduler.Default);行代码什么也不做。

然后返回原始的未修改的ole可观察序列并订阅它。

我有两个建议

创建一个Logging helper操作符来删除代码中的所有Console.Write噪声。在调试Rx代码时，这将是一个有用的东西，也将允许您学习如何创建Rx操作符(这将是无副作用的)。https://github.com/LeeCampbell/RxCookbook/blob/34ac4f3536b00bbe259384d3bf0e8746da3311cc/Instrumentation/Logging.md
1. 链接你的方法，这样你就能得到你想要的行为。

这里我们正确地使用了SubscribeOn方法。我们还使用Log扩展方法，您可以从上面的链接自己编写。

var ole = Observable.Create<string>(obs=>
{       
    Thread.Sleep(5000); //Dont use Thread.Sleep and Rx :-)
    obs.OnNext("point1");
    obs.OnCompleted();
    return Disposable.Empty;
});
var oleSubscription = ole
    .Log("ole")
    .SubscribeOn(NewThreadScheduler.Default)
    .ObserveOn(/*What ever Android's UI Thread Scheduler is*/)
    .Subscribe(s => Console.WriteLine("result: " + s));

还请注意，我也添加了ObserveOn操作符。我强烈建议这两个方法的用户只在最终订阅者(可能是您的ViewModel?)上使用它们，并且只将它们用作上述Subscribe方法之前的最后一个操作符。

更多帮助:

• http://www.introtorx.com/content/v1.0.10621.0/15_SchedulingAndThreading.html
• http://www.introtorx.com/content/v1.0.10621.0/18_UsageGuidelines.html UsageGuidelines