系列主题：基于消息的软件架构模型演变

 

一、反应式编程(Reactive Programming)

1、什么是反应式编程：反应式编程(Reactive programming)简称Rx，他是一个使用LINQ风格编写基于观察者模式的异步编程模型。简单点说Rx = Observables + LINQ + Schedulers。

2、为什么会产生这种风格的编程模型？我在本系列文章开始的时候说过一个使用事件的例子：

var watch = new FileSystemWatcher(); watch.Created += (s, e) => { var fileType = Path.GetExtension(e.FullPath); if (fileType.ToLower() == "jpg") { //do some thing } };

这个代码定义了一个FileSystemWatcher，然后在Watcher事件上注册了一个匿名函数。事件的使用是一种命令式代码风格，有没有办法写出声明性更强的代码风格？我们知道使用高阶函数可以让代码更具声明性，整个LINQ扩展就是一个高阶函数库，常见的LINQ风格代码如下：

var list = Enumerable.Range(1, 10) .Where(x => x > 8) .Select(x => x.ToString()) .First();

能否使用这样的风格来编写事件呢？

3、事件流
LINQ是对IEnumerable<T>的一系列扩展方法，我们可以简单的将IEnumerable<T>认为是一个集合。当我们将事件放在一个时间范围内，事件也变成了集合。我们可以将这个事件集合理解为事件流。

事件流的出现给了我们一个能够对事件进行LINQ操作的灵感。

二、反应式编程中的两个重要类型

事件模型从本质上来说是观察者模式，所以IObservable<T>和IObserver<T>也是该模型的重头戏。让我们来看看这两个接口的定义：

public interface IObservable<out T> { //Notifies the provider that an observer is to receive notifications. IDisposable Subscribe(IObserver<T> observer); } public interface IObserver<in T> { //Notifies the observer that the provider has finished sending push-based notifications. void OnCompleted(); //Notifies the observer that the provider has experienced an error condition. void OnError(Exception error); //Provides the observer with new data. void OnNext(T value); }

这两个名称准确的反应出了它两的职责：IObservable<T>-可观察的事物，IObserver<T>-观察者。显然事件流是可观察的事物，我们用Rx改写上面的例子：

Observable.FromEventPattern<FileSystemEventArgs>(watch, "Created") .Where(e => Path.GetExtension(e.EventArgs.FullPath).ToLower() == "jpg") .Subscribe(e => { //do some thing });

注：在.net下使用Rx编程需要安装以下Nuget组件：

Rx-Core

Rx-Interfaces

Rx-Linq

三、UI编程中使用Rx

Rx模型不但使得代码更加具有声明性，而且使得UI编程更加简单。

1、UI编程中的第一段Rx代码

为了简单的展示如何在UI编程中使用Rx，我们以Winform中的Button为例，看看事件模型和Rx有何不同。

private void BindFirstGroupButtons() { btnFirstEventMode.Click += btnFirstEventMode_Click; } void btnFirstEventMode_Click(object sender, EventArgs e) { MessageBox.Show("hello world"); }

添加了一个Button，点击Button的时候弹出一个对话框。使用Rx做同样的实现：

//得到了Button的Click事件流。 var clickedStream = Observable.FromEventPattern<EventArgs>(btnFirstReactiveMode, "Click"); //在事件流上注册了一个观察者。 clickedStream.Subscribe(e => MessageBox.Show("Hello world"));

有朋友指出字符串“Click”非常让人不爽，这确实是个问题。由于Click是一个event类型，无法用表达式树获取其名称，最终我想到使用扩展方法来实现：

public static IObservable<EventPattern<EventArgs>> FromClickEventPattern(this Button button) { return Observable.FromEventPattern<EventArgs>(button, "Click"); } public static IObservable<EventPattern<EventArgs>> FromDoubleClickEventPattern(this Button button) { return Observable.FromEventPattern<EventArgs>(button, "DoubleClick"); }

我们平时常用的事件类型也就那么几个，可以暂时通过这种方案来实现，该方案算不上完美，但是比起直接使用字符串又能优雅不少。

btnFirstReactiveMode.FromClickEventPattern() .Subscribe(e => MessageBox.Show("hello world"));

2、UI编程中存在一个很常见的场景：当一个事件的注册者阻塞了线程时，整个界面都处于假死状态。.net中的异步模型也从APM，EAP，TPL不断演化直至async/await模型的出现才使得异步编程更加简单易用。我们来看看界面假死的代码：

void btnSecondEventMode_Click(object sender, EventArgs e) { btnSecondEventMode.BackColor = Color.Coral; Thread.Sleep(5000); MessageBox.Show("hello world"); }

Thread.Sleep(5000);模拟了一个长时间的操作，当你点下Button时整个界面处于假死状态并且此时的程序无法响应其他的界面事件。传统的解决方案是使用多线程来解决假死：

void BtnSecondEventAsyncModel_Click(object sender, EventArgs e) { Action action = () => { Task.Run(() => { BtnSecondEventAsyncModel.BackColor = Color.Coral; Thread.Sleep(5000); MessageBox.Show("hello world"); }); }; BeginInvoke(action); }

这个代码的复杂点在于：普通的多线程无法对UI进行操作，在Winform中需要用Control.BeginInvoke(Action action)经过包装后，多线程中的UI操作才能正确执行，WPF则要使用Dispatcher.BeginInvoke(Action action)包装。在Rx中选择合适的Scheduler类型就可以轻松解决多线程更新UI的问题。

btnSecondReactiveMode.FromClickEventPattern() .ObserveOn(new NewThreadScheduler()) .Subscribe(e => { btnSecondReactiveMode.BackColor = Color.Coral; Thread.Sleep(5000); MessageBox.Show("hello world"); });

一句ObserveOn(new NewThreadScheduler())成功让后面的观察者跑在了新线程中，并且避免了多线程更新UI的问题，使用者无需再被这种问题所困扰。
注：使用Scheduler需要从Nuget中安装Rx-PlatformServices。

常用的Scheduler有：

DefaultScheduler:默认的Scheduler，会将观察者添加在调用队列中

ImmediateScheduler：立即执行

TaskPoolScheduler：用来执行短时间的任务

...

3、再来一个例子，让我们感受一下Rx的魅力