《C#并发编程经典实例》—— 发送通知给上下文

声明：本文是《c#并发编程经典实例》的样章，感谢图灵授权并发编程网站发布样章，禁止以任何形式转载此文。

问题

rx 尽量做到了线程不可知（thread agnostic）。因此它会在任意一个活动线程中发出通知（例如 onnext）。

但是我们通常希望通知只发给特定的上下文。例如 ui 元素只能被它所属的 ui 线程控制， 因此，如果要根据 rx 的通知来修改 ui，就应该把通知“转移”到 ui 线程。

解决方案

rx 提供了 observeon 操作符，用来把通知转移到其他线程调度器。 看下面的例子，使用 interval，每秒钟产生一个 onnext 通知：

用我的电脑测试，显示结果为：

ui thread is 9

interval 0 on thread 10

interval 1 on thread 10

interval 2 on thread 11

interval 3 on thread 11

interval 4 on thread 10

interval 5 on thread 11

interval 6 on thread 11

因为 interval 基于一个定时器（没有指定的线程），通知会在线程池线程中引发，而不是 在 ui 线程中。要更新 ui 元素，可以通过 observeon 输送通知，并传递一个代表 ui 线程 的同步上下文：

observeon 的另一个常用功能是可以在必要时离开 ui 线程。假设有这样的情况：鼠标一移

动，就意味着需要进行一些 cpu 密集型的计算。默认情况下，所有的鼠标移动事件都发 生在 ui 线程，因此可以使用 observeon 把通知移动到一个线程池线程，在那里进行计算， 然后再把表示结果的通知返回给 ui 线程：

private void button_click(object sender, routedeventargs e)

{

var uicontext = synchronizationcontext.current;

trace.writeline(“ui thread is ” + environment.currentmanagedthreadid); observable.fromeventpattern<mouseeventhandler, mouseeventargs>(

handler => (s, a) => handler(s, a), handler => mousemove += handler, handler => mousemove -= handler)

.select(evt => evt.eventargs.getposition(this))

.observeon(scheduler.default)

.select(position =>

// 复杂的计算过程。

thread.sleep(100);

var result = position.x + position.y; trace.writeline(“calculated result ” + result + ” on thread ” +

environment.currentmanagedthreadid);

return result;

})

.observeon(uicontext)

.subscribe(x => trace.writeline(“result ” + x + ” on thread ” + environment.currentmanagedthreadid));

}

运行这段代码的话，就会发现计算过程是在线程池线程中进行的，计算结果在 ui 线程中

显示。另外，还会发现计算和结果会滞后于输入，形成等待的队列，这种现象出现的原因 在于，比起 100 秒 1 次的计算，鼠标移动的更新频率更高。rx 中有几种技术可以处理这 种情况，其中一个常用方法是对输入流速进行限制，具体会在 5.4 节介绍。

讨论

实际上，observeon 是把通知转移到一个 rx 调度器上了。本节介绍了默认调度器（即线程 池）和一种创建 ui 调度器的方法。observeon 最常用的功能是移到或移出 ui 线程，但调 度器也能用于别的场合。6.6 节介绍高级测试时，将再次关注调度器。