.NET 响应式编程 System.Reactive 系列文章（一）：基础概念
.NET 响应式编程 System.Reactive 系列文章（一）：基础概念
引言
在现代软件开发中，处理
异步事件
和
数据流
已经成为常见的需求，比如用户输入、网络请求、传感器数据等。这些数据流通常是
无限的、异步的、实时的
，而传统的编程方式往往无法优雅地处理这些情况。
响应式编程(Reactive Programming)
为我们提供了一种新的思路，帮助开发者更自然、更高效地管理数据流和异步事件。
在 .NET 中，响应式编程的核心库是
System.Reactive
，通常简称为
Rx
。本篇文章将介绍响应式编程的基础概念以及
System.Reactive
的核心组件，为后续深入学习奠定基础。
什么是响应式编程？
响应式编程（Reactive Programming）
是一种
声明式编程范式
，专注于
异步数据流
和
变化传播
。简单来说，它是一种处理
事件驱动
和
数据变化
的编程方式，可以让程序自动对外部的变化做出反应。
在响应式编程中：
- 数据流可以是
有界的
或
无界的
（无限的）。 - 数据流的变化可以触发
订阅者
的行为。 - 订阅者（Observer）可以随时
订阅
或
取消订阅
这些数据流。
传统编程 vs. 响应式编程
| 传统编程 | 响应式编程 |
|---|---|
| 通过轮询来获取数据变化 | 自动响应数据流的变化 |
| 使用回调函数处理异步 | 通过订阅和流式操作符处理异步 |
| 不擅长处理无限数据流 | 专注于处理无限、异步的数据流 |
System.Reactive 概述
System.Reactive
是微软推出的
Reactive Extensions（Rx）
的实现，为 .NET 提供了一个强大的
观察者模式
和
操作符库
，让我们可以轻松地管理数据流和异步事件。
核心组件
| 组件 | 描述 |
|---|---|
IObservable<T>
|
表示一个数据流的 生产者 |
IObserver<T>
|
表示一个数据流的 消费者 （订阅者） |
Subject<T>
|
既是 生产者 也是 消费者 |
| 操作符（Operators） | 用于对数据流进行转换、过滤、组合等操作 |
观察者模式简介
System.Reactive
的核心是基于
观察者模式（Observer Pattern）
，这是一种常见的设计模式，广泛用于处理事件和回调。
观察者模式的核心接口
IObservable
（可观察对象）
- 负责
生产
数据流。 - 提供
Subscribe
方法，允许观察者订阅它的数据流。
- 负责
IObserver
（观察者）
- 负责
消费
数据流。 - 定义了以下三个方法：
OnNext(T value)
: 当有新数据时被调用。OnError(Exception error)
: 当数据流发生错误时被调用。OnCompleted()
: 当数据流结束时被调用。
- 负责
简单的示例代码
using System;
using System.Reactive.Subjects;
public class Program
{
public static void Main()
{
// 创建一个 Subject，它既是 IObservable 也是 IObserver
var subject = new Subject<string>();
// 订阅数据流
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();
}
}
输出结果：
Received: Hello
Received: Reactive Extensions
Completed
Observable vs. Task
许多人会将
Observable
和
Task
进行比较，因为它们都用于处理异步操作。但两者之间有一些显著的区别。
| 特性 | Observable | Task |
|---|---|---|
| 数据流 | 多个值 / 无限值 | 单个值 |
| 生命周期 | 可被取消订阅 | 一次性操作 |
| 时间维度 | 持续的时间序列 | 单次完成的任务 |
| 支持的操作符 | 丰富的转换、过滤、组合操作符 | 少数操作符 |
简单总结：
- Task
更适合处理
单次异步操作
。 - Observable
更适合处理
连续的数据流
或
多次异步事件
。
数据流的三个阶段
在响应式编程中，数据流有三个阶段：
- OnNext
: 数据流的每一个值都会通过
OnNext
方法传递给订阅者。 - OnError
: 如果数据流中出现错误，会通过
OnError
方法通知订阅者。 - OnCompleted
: 当数据流结束时，会通过
OnCompleted
方法通知订阅者。
热数据流和冷数据流
在
System.Reactive
中，数据流可以分为两种类型：
1. 冷数据流（Cold Observable）
- 冷数据流是
被订阅时才开始产生数据
。 - 每个订阅者都会从
头开始
接收数据。
示例：
var cold = Observable.Range(1, 5);
cold.Subscribe(x => Console.WriteLine($"Subscriber 1: {x}"));
cold.Subscribe(x => Console.WriteLine($"Subscriber 2: {x}"));
输出：
Subscriber 1: 1
Subscriber 1: 2
Subscriber 1: 3
Subscriber 1: 4
Subscriber 1: 5
Subscriber 2: 1
Subscriber 2: 2
Subscriber 2: 3
Subscriber 2: 4
Subscriber 2: 5
2. 热数据流（Hot Observable）
- 热数据流是
数据流开始时就产生数据
。 - 每个订阅者会从
当前数据流的位置
开始接收数据。
示例：
var hot = new Subject<int>();
hot.OnNext(1);
hot.Subscribe(x => Console.WriteLine($"Subscriber: {x}"));
hot.OnNext(2);
输出：
Subscriber: 2
总结
在本篇文章中，我们介绍了响应式编程的基础概念以及
System.Reactive
的核心组件：
- 响应式编程专注于处理
异步数据流
。 - System.Reactive
提供了核心接口
IObservable
和
IObserver
。 - 数据流的生命周期包含
OnNext
、
OnError
和
OnCompleted
。 - 区分了
冷数据流
和
热数据流
。
下一篇文章将介绍
System.Reactive
的基础操作符，包括创建、转换和过滤数据流的方法，敬请期待！
