C#-并发编程之【四】响应式编程简介

C#并发编程之【四】响应式编程简介

原创MoMaek浅陌发布于2019-04-11 14:17:53 阅读数 188 收藏

一、什么是响应式编程

响应式编程可以像处理数据流一样处理事件流。根据经验，如果事件中带有参数，那么最好采用响应式编程，而不是常规的事件处理程序。
响应式编程基于“可观察的流”（observable stream）这一概念。你一旦申请了可观察流，就可以收到任意数量的数据项（OnNext），并且流在结束时会发出一个错误（OnError）或一个 “流结束”的通知（OnCompleted）。
有些可观察流是不会结束的。实际的接口就像这样：

interface IObserver<in T>     {
         void OnNext(T item);         
         void OnCompleted();        
          void OnError(Exception error);     
 } 
interface IObservable<out T>             {
	Disposable Subscribe(IObserver<T> observer); 
}

但是，我们通常不需要实现这些接口。因为 Reactive Extensions（Rx）库已经实现了所有接口。响应式编程的最终代码非常像 LINQ，可以认为它就是“LINQ to events”。

二、如何实现

LINQ 具有的特性，Rx 也都有。Rx 在此基础上增加了很多它自己的操作符，特别是与时间有关的操作符：

 Observable.Interval(TimeSpan.FromSeconds(1))  //延时一段时间的计数器（Interval）
 	.Timestamp() //为每个事件加 了一个时间戳（Timestamp）
 	 .Where(x => x.Value % 2 == 0)  //对事件进行过滤，只包含偶数值（Where）       
 	 .Select(x => x.Timestamp)  //选择时间 戳的值（Timestamp）      
 	  .Subscribe(x => Trace.WriteLine(x)); //当每个时间戳值到达时，把它输入调试器（Subscribe）

主要区别在于：LINQ to Object 和 LINQ to Entity 使用“拉取”模式，LINQ 的枚举通过查询拉出数据。而 LINQ to event （Rx）使用“推送”模式，事件到达后就自行穿过查询。

可观察流的定义和其订阅是互相独立的。上面最后一个例子与下面的代码等效：

IObservable<DateTimeOffset> timestamps =  
Observable.Interval(TimeSpan.FromSeconds(1))         
	.Timestamp()         
	.Where(x => x.Value % 2 == 0)         
	.Select(x => x.Timestamp);     
	timestamps.Subscribe(x => Trace.WriteLine(x));

一种常规的做法是把可观察流定义为一种类型，然后将其作为 IObservable< T > 资源使用。其他类型可以订阅这些流，或者把这些流与其他操作符组合，创建另一个可观察流。
Rx 的订阅也是一个资源。Subscribe 操作符返回一个 IDisposable，即表示订阅完成。当你响应了那个可观察流，就得处理这个订阅。