有关Reactive Extensions的介绍可见https://rx.codeplex.com/，总的来说，你可以当它是又一个异步编程的框架，它以观察者模式实现了对数据流的的“订阅”。一个列表，一个事件，一个耗时操作的方法，等等，都可以Observe起来，并且注册它的变动（任何变动将调用IObservable的OnNext/OnError/OnCompleted方法），这就大大统一的异步编程的编程方式；同时所有Observable的对象还可以任意组合；最后，支持用Linq语法来进行投影，将各种数据流转化为你感兴趣的数据对象，总之，是值得去了解一下的。不但如此，微软还为js, python, c++, ruby各做了一套同样理念的实现，诚意满满，自信满满啊。
学习linq有101sample，也就有人为Rx做了一个101sample的网站：http://rxwiki.wikidot.com/101samples
入门示例感兴趣的可以对着101samples做，今天要演示的是rx与c#5.0引入的Task和Async来做一个编程方式和效率上的简单对比。其实task的引入已经大大方便了异步编程，不再像传统的异步编程一样要写很多丑陋的begin/end了，但是它更像的是语法糖，而Rx则做的是整合。
说一下需求，冷冷姑娘对上猇产生了兴趣，想把她的博客全部down下来，我分析了一下，这个163博客既没有提供rss全文输出，rss也只能输出到最近40条。如果从网页里翻页截取，他们又用js把请求给包装起来了，算了吧，我还是偷点懒，直接从rss里读出那40篇文章的链接，然后异步下载回来吧！这就是原始需求，现在用两种写法来干这件事。
1，先看看我怎么把rss解析出来，.net有自己的类
using (var reader = XmlReader.Create(url))
{
var feed = System.ServiceModel.Syndication.SyndicationFeed.Load(reader);
//feed已经取出来了，rss2.0标准，已经返成了一个对象
//your code here
}
2, Task的写法
private async Task run()
{
using (var reader = XmlReader.Create(url))
{
var result = feed.Items.Select(m => new { date = m.PublishDate, link = m.Links[0].Uri.ToString() });
foreach (var item in result)
{
try
{
var html = await new HttpClient().GetStringAsync(item.link);
Console.WriteLine(parse(html, item.date));
}
catch { }
}
}
}
3，Rx的写法
前面说过，任何事件/网络流/文件操作，都可以转化成一个Observable的对象
private void runrx()
{
var sw = new Stopwatch();
sw.Start();
using (var reader = XmlReader.Create(url))
{
var feed = SyndicationFeed.Load(reader);
//所以fee.Items可以ToObservable
var data = feed.Items.ToObservable()
.Select(item => new
{
//同样，异步任务也可以ToObservable
txt = new HttpClient().GetStringAsync(item.Links[0].Uri.ToString()).ToObservable(),
pubtime = item.PublishDate
});//投影成了一个IObservable和一个DateTime组成的数据流
data.Subscribe(d => d.txt.Subscribe(t => Console.WriteLine(parse(t, d.pubtime)),//OnSuccess
e => Console.WriteLine(e.Message), //OnError
() => Console.WriteLine(sw.ElapsedMilliseconds) //OnCompleted
));
Console.ReadKey();
sw.Stop();
}
}
4，至于里面用到了parse方法，那是用正则在提取网页内容
这个其实可以不演示的，如果从163博客里导文章这种低概率事件正好也被你碰上了的话，下面给个参考：
static string parse(string source, DateTimeOffset date)
{
try
{
Regex r = new Regex(@"<span class=""tcnt"">([^<]*)</span>[\s\S]*<div class=""nbw-blog-start""></div>([\s\S]*)<div class=""nbw-blog-end""></div>");
var g = r.Matches(source)[0].Groups;
//return new { title = g[1].Value, content = g[2].Value };
return
string.Format(
"<!doctype html><html><head><meta charset='UTF-8'><title>{0}</title></head><body><h1>{0}</h1><p>{2}</p>{1}</body>",
g[1].Value, g[2].Value, date.ToString("yyyy-MM-dd"));
}
catch (Exception)
{
return "";
}
}
5，在控制台里调用：
private const string url = "http://blog.163.com/ttoonnyy_2006/rss";
private static void Main(string[] args)
{
var p = new Program();
//只要有参数传进来就调Rx进行测试
if (args.Length > 0)
{
p.runrx();
}
else
{
var sw = new Stopwatch();
sw.Start();
var t = p.run();
t.Wait();
sw.Stop();
Console.WriteLine(sw.ElapsedMilliseconds);
}
}
当然这些代码是后来为了测试用，完成别人所托只需要在取得字符串后异步存到文档里，顺便用reveal这样的演示插件把它弄漂亮点就行了。
我做了Stopwatch，看看结果对比吧。
测试从RSS解析出所有的文章链接（40篇），然后异步下载，输出到控制台的总时间，测试十次，结果如下，单位是毫秒：
rx task
1377 6625
1087 6678
1108 8207
1404 8647
1261 7463
1158 6876
1253 6563
1244 6656
1058 7752
1107 6230
差距还是很明显的，async这种写法上异步是异步了，但是却要await，等于是分头取，然后挨个提交，表现是什么呢？如果把上文中的日志时间打印出来，它的顺序是跟日志的产生时间一致的，而Rx而是典型的一起取一起提交，谁在前完全取决于谁快了。所以它每次的顺序是不一致的。
此外，纯语法来说，task/async写法还是门槛为0，代码也简洁很多，Rx胜在他的整合性，比如上面的例子，就把一个异步的流和一个feed里面的items给整合成了一个新的Observable对象，这种组合是有无穷想象力的。
例子很简单，有兴趣的可以在自己的编程过程中有意识地对二者做个对比。
6，问题
上例中，我没找到在Rx事件结束的时候来停止stopwatch的方法，所以在每一个subscribe的OnComplete事件里面计了次时，献计献策？