Reactive Extensions(Rx) 学习

Bruce Eckel（著有多部编程书籍）和Jonas Boner（Akka的缔造者和Typesafe的CTO）发表了“反应性宣言”，在其中尝试着定义什么是反应性应用。

这样的应用应该能够：

对事件做出反应：事件驱动的本质，让反应性应用能够支持文中提到的若干特性。
对负载做出反应：聚焦于可扩展性，而不是单用户性能。
对失败做出反应：建立弹性系统，能够从各个层级进行恢复。
对用户做出反应：综合上述特征，实现交互式用户体验。

在这份宣言公布之后，Scala的创造者Martin Odersky、Reactive Extensions的创造者Erik Meijer和Akka科技公司的领导者Roland Kuhn，在Coursera上发布了一套免费课程，名为“反应性编程原理”：

该课程的目标在于讲授反应性编程的原理。反应性编程是一门新兴的学科，结合了并发、事件驱动和异步系统。对于编写任何类型的Web服务或分布式系统来说，它都至关重要；同时它在众多高性能并发系统中占有核心位置。反应性变成可以被视作高阶函数式编程对并发系统的自然拓展，通过协调和编排Actor交换的异步数据流，来处理分布的状态。

Reactive Extensions(Rx)的优点在于能够将传统的异步编程方式从支离破碎的代码调用中解放出来。Rx能够使的我们可以将异步代码写到一个单独的方法中，使得代码可读性和可维护性大大增强。

《Reactive Extensions介绍》我们了解了Rx中的一些比较重要的操作符，本文中我们将会学习如何将Reactive Extensions(Rx)应用到我们的应用程序中。

同步方法调用是阻塞式的，在很多场景下这是不合适的。我们能够用Rx改造成异步调用。一个最简单的方法就是使用IObservable.Start方法，使得Rx为我们来管理这些异步调用。

    public static void ObservableStart(int x, int y)

        {

            PlusTwoNumberAsync(x, y).Subscribe(Console.WriteLine);

            Console.ReadKey();

        }

        private static IObservable<int> PlusTwoNumberAsync(int x, int y)

        {

            return Observable.Start(() => PlusTwoNumber(x, y));

        }

        private static int PlusTwoNumber(int x, int y)

        {

            Thread.Sleep(5000);

            return x + y;

        }

除了Observable.Start外也可以使用Observable.Return来将同步方法改造为异步方法。只需要将上面的PlusTwoNumberAsync方法改为下面即可，运行程序的效果相同。

     private static IObservable<int> PlusTwoNumberReturnAsync(int x, int y)

        {

            return Observable.Return(PlusTwoNumber(x, y));

        }

使用SelectMany可以很方便的实现诸如在一个异步方法中调用另外一个异步方法的功能。

public static void ObservableSelectMany(int x, int y)
{

PlusTwoNumberStartAsync(x, y).SelectMany(aPlusB => MultiplyByFiveAsync(aPlusB)).Subscribe(Console.WriteLine);

}

private static IObservable<int> MultiplyByFiveAsync(int x)

{

return Observable.Return(MultiplyByFive(x));

}

private static int MultiplyByFive(int x)

{

Thread.Sleep(5000);

return x * 5;

}

完整代码如下：

// -----------------------------------------------------------------------

// <copyright file="RxAsyncCall.cs" company="">

// TODO: Update copyright text.

// </copyright>

// -----------------------------------------------------------------------

namespace RxPractice

{

    using System;

    using System.Collections.Generic;

    using System.Linq;

    using System.Text;

    using System.Reactive.Linq;

    using System.Threading;

    /// <summary>

    /// 异步调用

    /// </summary>

    public class RxAsyncCall

    {

        public static void ObservableStart(int x, int y)

        {

            PlusTwoNumberStartAsync(x, y).Subscribe(Console.WriteLine);

        }

        public static void ObservableReturn(int x, int y)

        {

            PlusTwoNumberReturnAsync(x, y).Subscribe(Console.WriteLine);

        }

        public static void ObservableSelectMany(int x, int y)

        {

            PlusTwoNumberStartAsync(x, y).SelectMany(aPlusB => MultiplyByFiveAsync(aPlusB)).Subscribe(Console.WriteLine);

        }

        private static IObservable<int> PlusTwoNumberStartAsync(int x, int y)

        {

            return Observable.Start(() => PlusTwoNumber(x, y));

        }

        private static int PlusTwoNumber(int x, int y)

        {

            Thread.Sleep(2000);

            return x + y;

        }

        private static IObservable<int> MultiplyByFiveAsync(int x)

        {

            return Observable.Return(MultiplyByFive(x));

        }

        private static int MultiplyByFive(int x)

        {

            Thread.Sleep(5000);

            return x * 5;

        }

        private static IObservable<int> PlusTwoNumberReturnAsync(int x, int y)

        {

            return Observable.Return(PlusTwoNumber(x, y));

        }

    }

}