在上篇文章《Reactive Extensions简介一》里，我简单的介绍了一下观察者模式，并给出了一个用Rx简化观察者模式的示例。今天我在这里开始介绍一下Rx里创建Iobservables序列的几种常用方法。
1. 基本序列
IObservable<int> obs = Observable.Empty<int>(); //空序列
IObservable<int> obs = Observable.Return(8); //包含一个元素8的序列
IObservable<int> obs = Observable.Throw<int>(new Exception()); //一个产生异常的序列
2. 简单序列
IObservable<int> obs = Observable.Repeat(5, 3); //生成包含一个重复值的序列
这两个函数与Enumerable.Repeat和Enumerable.Range类似，这里就不多说明了。
3. 周期序列
IObservable<long> obs = Observable.Interval(TimeSpan.FromSeconds(1));
IObservable<long> obs = Observable.Timer(TimeSpan.FromMinutes(1), TimeSpan.FromSeconds(1));
这两个函数根据参数制定的时间间隔产生一个周期序列，其值从0开始，每次通知时其值递增1。（感觉这两个函数功能是一样的，不知道为什么不把它们合成一个函数）例如，对于如下函数：
Observable.Interval(TimeSpan.FromSeconds(1))
.Subscribe(i => Console.WriteLine(i + " >>> " + DateTime.Now));
其输出结果为
0 >>> 2011/5/2 21:15:36
1 >>> 2011/5/2 21:15:37
2 >>> 2011/5/2 21:15:38
… …
4. 从IEnumerable中生成
IObservable<int> obs = new int[] { 1, 2, 3, 4 }.ToObservable();
同样，也可以通过Observable.ToEnumerable函数，将IObservable对象转换为IEnumerable对象。（同步序列单元测试更容易）
5. 从异步运算中生成
可以将异步运算结果封装为了一个元素个数为1的IObservable序列中。
//异步处理
Observable.Start(() => 1);
//Task
Task.Factory.StartNew(...).ToObservable();
//异步函数
//典型的异步处理为异步IO及WebService
Func<int, int, double> sampleFunc = (a, b) => 1d;
Func<int, int, IObservable<double>> funcObs = Observable.FromAsyncPattern<int, int, double>(sampleFunc.BeginInvoke, sampleFunc.EndInvoke);
IObservable<double> obs = funcObs(1, 0);
如果要将连续的异步运算结果封装起来，则还需要用到Rx库的其它函数，这个后面再介绍。
6. 从事件中生成
可以通过Observable.FromEvent函数将事件封装为一个IObservable序列：
var txt = new TextBox();
var input = Observable.FromEvent<EventArgs>(txt, "TextChanged")
这个函数有好几种重载形式，这个是我最喜欢的一种。但用到了反射，性能党可自行选择其它形式。
7. Observable.Generate函数
Observable.Generate通过一种迭代的形式生成序列，类似for循环，是一种比较常用的方式：
IObservable<int> obs = Observable.Generate(
0, //初始状态
i => i < 5, //终止条件
i => i + 1, //迭代条件
i => i * i); //结果
另外，还有一个Observable.GenerateWithTime函数，可以指定产生的时间间隔。使用方式非常类似，这里就不介绍了。
8. Observable.Create函数
Observable.Create函数是通过主动调用OnNext、OnError及OnCompleted函数进行通知从而生成IObservable序列。
IObservable<int> obs = Observable.Create<int>(o =>
{
o.OnNext(1);
o.OnNext(2);
o.OnCompleted();
return () => Console.WriteLine("Finished");
});
该函数返回一个Function，会在序列通知完成后执行，一般用于执行一些析构的操作。另外，与之类似的还有一个Observable.CreateWithDisposable函数，返回的是一个IDispose对象，序列通知完成后会执行它的Dispose操作。
与Observable.Generate比起来，Observable.Create更加灵活。例如前面的例子可以改写如下：
IObservable<int> obs = Observable.Create<int>(o =>
{
for (int i = 0; i < 5; i++)
{
o.OnNext(i * i);
}
o.OnCompleted();
return () => { };
});
Observable.Create及Observable.CreateWithDisposable两个函数较常见的用法是和Subscribe函数一起联合使用，从一个序列衍生出另一个序列。具体的例子可以参考我前面《Reactive Extensions学习》文章中的例子。
9. Subject对象
前面我们介绍的都是一种自动通知的IObservable序列，很多时候，我们还需要一种手动通知的IObservable序列。例如，对于这样一个功能：当某文本框里面的数字改变时发送一次数据通知，同时每点击一次发送按钮可以重新发送一次数据通知。这用前面的方法就无法实现了。
一个解决的办法是类似我前文里那样自己实现一个IObservable对象，然后开放数据通知接口，文本改变或点击按钮时主动调用该接口发送通知。实现这个功能并不难，不过Rx已经为我们实现了一个名为Subject的类，我们就不用造重复轮子了。
var obs = new Subject<int>();
obs.Subscribe(Console.WriteLine);
for (int i = 0; i < 5; i++)
{
obs.OnNext(i * i);
}
obs.OnCompleted();