Rx.NET

编程范式：

编程范式分为命令式编程、声明式编程、函数式编程和响应式编程。

命令式编程： 命令式编程的主要思想是关注计算机执行的步骤，即一步一步告诉计算机先做什么再做什么。

//1. 声明变量
List<int> results = new List<int>();
//2. 循环变量
foreach(var num in Enumerable.Range(1,10))
{
    //3. 添加条件
    if (num > 5)
    {
        //4. 添加处理逻辑
        results.Add(num);
        Console.WriteLine(num);
    }
}

声明式编程： 声明式编程是以数据结构的形式来表达程序执行的逻辑。它的主要思想是告诉计算机应该做什么，但不指定具体要怎么做。

var nums = from num in Enumerable.Range(1,10) where num > 5 select num

函数式编程： 主要思想是把运算过程尽量写成一系列嵌套的函数调用。

Enumerable.Range(1, 10).Where(num => num > 5).ToList().ForEach(Console.WriteLine);

响应式编程： 响应式编程是一种面向数据流和变化传播的编程范式，旨在简化事件驱动应用的实现。响应式编程专注于如何创建依赖于变更的数据流并对变化做出响应。

IObservable<int> nums = Enumerable.Range(1, 10).ToObservable();

IDisposable subscription = nums.Where(num => num > 5).Subscribe(Console.WriteLine);

subscription.Dispose();

Rx.NET

Reactive Extensions（Rx）是一个为.NET应用提供响应式编程模型的库，用来构建异步基于事件流的应用，通过安装System.ReactiveNuget包进行引用。Rx将事件流抽象为Observable sequences（可观察序列）表示异步数据流，使用LINQ运算符查询异步数据流，并使用Scheduler来控制异步数据流中的并发性。简单地说：Rx = Observables + LINQ + Schedulers。

Rx.NET核心：

一切皆为数据流
Observable是对数据流的抽象（事件源）
Observer是对Observable的响应（事件处理程序）

在Rx中，分别使用IObservable<T>和IObserver<T>接口来表示可观察序列和观察者。

IObservable
（可继承）

public interface IObservable<out T>
{
      //Notifies the provider that an observer is to receive notifications.
      IDisposable Subscribe(IObserver<T> observer);
}

IObserver
（可继承）

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方式（事件源）：

直接实现IObservable<T>接口

private class MyObservable : IObservable<int>
{
    public IDisposable Subscribe(IObserver<int> observer)
    {
        for (int i = 0; i < 100; i++)
        {
            observer.OnNext(i);
        }
        observer.OnCompleted();
        return Disposable.Empty;
    }
}

使用Observable.Create创建

var observable = Observable.Create<int>(observer =>
 {
     for (int i = 0; i < 100; i++)
     {
         observer.OnNext(i);
     }
     observer.OnCompleted();
     return Disposable.Empty;
 });

使用Observable.Deffer进行延迟创建（当有观察者订阅时才创建）

Observable.Defer(() =>
{
    var connection = MysqlConnection(user, password);
    return connection.ToObservable();
});

比如要连接数据库进行查询，如果没有观察者，那么数据库连接会一直被占用，这样会造成资源浪费。使用Deffer可以解决这个问题。 4. 使用Observable.Generate创建迭代类型的可观察序列

// 参数说明：
// initial state(初始值): 0
// condition (false means terminate)(条件(false表示终止)):i < 10
// next iteration step(下一个迭代步骤):i+1
// the value in each iteration(每个迭代中的值):i * 3
// 约等于:
// for (int i = 0; i < 10; i+1)
// {
//   yeild return i * 3;
// }
IObservable<int> observable = Observable.Generate(0, i => i < 10, i => i+1, i => i * 3);

使用Observable.Range创建指定区间的可观察序列

IObservable<int> observable = Observable.Range (0, 10).Select (i => i * 2);

特殊序列

Observable.Return ("Hello World");//创建单个元素的可观察序列
Observable.Never<string> ();//创建一个空的永远不会结束的可观察序列
Observable.Throw<ApplicationException> (new ApplicationException ("something bad happened"))//创建一个抛出指定异常的可观察序列
Observable.Empty<string> ()//创建一个空的立即结束的可观察序列

使用ToObservable转换IEnumerate和Task类型

Enumerable.Range(1, 10).ToObservable();
IObservable<IEnumerable<string>> resultsA = searchEngineA.SearchAsync(term).ToObservable();

使用Observable.Using进行资源释放

IObservable<string> lines = Observable.Using (() =>
// opens the file and returns the stream we work with
File.OpenText ("TextFile.txt"),
        stream =>
        Observable.Generate (
            stream, //initial state
            s => !s.EndOfStream, //we continue until we reach the end of the file
            s => s, //the stream is our state, it holds the position in the file
            s => s.ReadLine ()) //each iteration will emit the current line (and moves to the next)
    );

使用Observable.Interval创建指定间隔可观察序列

Observable.Interval(TimeSpan.FromMilliseconds(1));

使用Observable.Timer创建可观察的计时器

Observable.Timer(TimeSpan.FromMilliseconds(2), TimeSpan.FromMilliseconds(1));

实现IObserver方式（事件处理程序）：

继承IObserver

// 创建Observer
public class MyServer : IObserver<int>
{
    //序列完成通知
    public void OnCompleted()
    {
        Console.WriteLine("完成");
    }
    //异常
    public void OnError(Exception error)
    {
        Console.WriteLine(error.Message);
    }
    //提供数据(out)
    public void OnNext(int value)
    {
        Console.WriteLine(value);
    }
}

//使用Observer
subject.Subscribe(new MyServer());

通过Observer.Create()创建

// 创建Observer:
// Observer.Create<int>(num => Console.WriteLine(num))

// 使用Observer：
subject.Subscribe(Observer.Create<int>(num => Console.WriteLine(num)));

直接用Lambda表达式表示：

subject.Subscribe(num => Console.WriteLine(num));

Scheduler(控制并发,程序调度)

Rx提供了以下几种Scheduler：

NewThreadScheduler：即在新线程上执行
ThreadPoolScheduler：即在线程池中执行
TaskPoolScheduler：与ThreadPoolScheduler类似
CurrentThreadScheduler：在当前线程执行
ImmediateScheduler：在当前线程立即执行
EventLoopScheduler：创建一个后台线程按序执行所有操作

示例：

Observable.Return("Hello",NewThreadScheduler.Default)
.Subscribe(str=>Console.WriteLine($"{str} on ThreadId：{Thread.CurrentThread.ManagedThreadId}")
);
Console.WriteLine($"Current ThreadId：{Thread.CurrentThread.ManagedThreadId}");

// 以上输出：
// Current ThreadId：1
// Hello on ThreadId：4

一般调用流程

申明Subject
创建数据源Observable
（可观察序列）
创建处理函数Observer（观察者）并订阅subject
subject订阅数据源observable

//申明Subject（Subject<T>），借助Subject进行多播传输,注意，Subject内的事件是并行分发的
//也可以不用依靠Subject分发，但那是以文档流分发,例如通过事件源订阅：
//observable.Subscribe(i => Console.WriteLine(i));
Subject<int> subject = new Subject<int>();
//创建Observable<T>,可观察序列(数据源)
var observable = Observable.Create<int>(observer =>
{
    for (int i = 0; i < 100; i++)
    {
        observer.OnNext(i);
    }
    observer.OnCompleted();
    return Disposable.Empty;
});
//处理函数（观察者）订阅subject
subject.Subscribe(Observer.Create<int>(num => Console.WriteLine(num)));
subject.Subscribe(num => Console.WriteLine(num));
//最后subject订阅observable
observable.Subscribe(subject);

编程范式：#

Rx.NET#

Rx.NET核心：#

创建IObservable方式（事件源）：#

实现IObserver方式（事件处理程序）：#

Scheduler(控制并发,程序调度)#

一般调用流程#

编程范式：

Rx.NET

Rx.NET核心：

创建IObservable方式（事件源）：

实现IObserver方式（事件处理程序）：

Scheduler(控制并发,程序调度)

一般调用流程