.NET 响应式编程 System.Reactive 系列文章（一）：基础概念

摘要：在现代软件开发中，处理异步事件和数据流已经成为常见的需求，比如用户输入、网络请求、传感器数据等。这些数据流通常是无限的、异步的、实时的，而传统的编程方式往往无法优雅地处理这些情况。响应式编程(Reactive Programming)为我们提供了一种新的思路，

在现代软件开发中，处理异步事件和数据流已经成为常见的需求，比如用户输入、网络请求、传感器数据等。这些数据流通常是无限的、异步的、实时的，而传统的编程方式往往无法优雅地处理这些情况。响应式编程(Reactive Programming)为我们提供了一种新的思路，帮助开发者更自然、更高效地管理数据流和异步事件。

在 .NET 中，响应式编程的核心库是 System.Reactive，通常简称为Rx。本篇文章将介绍响应式编程的基础概念以及System.Reactive的核心组件，为后续深入学习奠定基础。

响应式编程（Reactive Programming）是一种声明式编程范式，专注于异步数据流和变化传播。简单来说，它是一种处理事件驱动和数据变化的编程方式，可以让程序自动对外部的变化做出反应。

在响应式编程中：

数据流可以是有界的或无界的（无限的）。

数据流的变化可以触发订阅者的行为。

订阅者（Observer）可以随时订阅或取消订阅这些数据流。

#传统编程 vs. 响应式编程

传统编程

响应式编程

通过轮询来获取数据变化

自动响应数据流的变化

使用回调函数处理异步

通过订阅和流式操作符处理异步

不擅长处理无限数据流

专注于处理无限、异步的数据流

#核心组件

组件

描述

IObservable

表示一个数据流的生产者

IObserver

表示一个数据流的消费者（订阅者）

Subject

既是生产者也是消费者

操作符（Operators）

用于对数据流进行转换、过滤、组合等操作

System.Reactive的核心是基于观察者模式（Observer Pattern），这是一种常见的设计模式，广泛用于处理事件和回调。

#观察者模式的核心接口

IObservable（可观察对象）

负责生产数据流。

提供 Subscribe方法，允许观察者订阅它的数据流。

IObserver（观察者）

OnNext(T value)

: 当有新数据时被调用。

OnError(Exception error)

: 当数据流发生错误时被调用。

OnCompleted

: 当数据流结束时被调用。

负责消费数据流。

定义了以上三个方法；

#简单的示例代码usingSystem;
usingSystem.Reactive.Subjects;

publicclassProgram
{
publicstaticvoidMain
{
// 创建一个 Subject，它既是 IObservable 也是 IObserver
varsubject=newSubjectstring>;

// 订阅数据流
subject.Subscribe(
onNext:value=>Console.WriteLine($"Received: {value}"),
onError: error=>Console.WriteLine($"Error: {error.Message}"),
onCompleted: =>Console.WriteLine("Completed")
);

// 发布数据
subject.OnNext("Hello");
subject.OnNext("Reactive Extensions");
subject.OnCompleted;
}
}

输出结果：

特性

Observable

Task

数据流

多个值 / 无限值

单个值

生命周期

可被取消订阅

一次性操作

时间维度

持续的时间序列

单次完成的任务

支持的操作符

丰富的转换、过滤、组合操作符

少数操作符

简单总结：

Task

更适合处理单次异步操作。

Observable

更适合处理连续的数据流或多次异步事件。

在响应式编程中，数据流有三个阶段：

OnNext

: 数据流的每一个值都会通过 OnNext方法传递给订阅者。

OnError

: 如果数据流中出现错误，会通过 OnError方法通知订阅者。

OnCompleted

: 当数据流结束时，会通过 OnCompleted方法通知订阅者。

在 System.Reactive中，数据流可以分为两种类型：

#1. 冷数据流（Cold Observable）

冷数据流是被订阅时才开始产生数据。

每个订阅者都会从头开始接收数据。

示例：

varcold=Observable.Range(1 , 5 );
cold.Subscribe(x=>Console.WriteLine($"Subscriber 1: {x}"));
cold.Subscribe(x=>Console.WriteLine($"Subscriber 2: {x}"));

输出：

Subscriber 1 :1
Subscriber1 :2
Subscriber1 :3
Subscriber1 :4
Subscriber1 :5
Subscriber2 :1
Subscriber2 :2
Subscriber2 :3
Subscriber2 :4
Subscriber2 :5
#2. 热数据流（Hot Observable）

热数据流是数据流开始时就产生数据。

每个订阅者会从当前数据流的位置开始接收数据。

示例：

varhot=newSubjectint>;
hot.OnNext(1 );
hot.Subscribe(x=>Console.WriteLine($"Subscriber: {x}"));
hot.OnNext(2 );

输出：

在本篇文章中，我们介绍了响应式编程的基础概念以及 System.Reactive的核心组件：

响应式编程专注于处理异步数据流。

System.Reactive

提供了核心接口 IObservable和IObserver。

数据流的生命周期包含 OnNext、OnError和OnCompleted。

区分了冷数据流和热数据流。

下一篇文章将介绍 System.Reactive的基础操作符，包括创建、转换和过滤数据流的方法，敬请期待！