C# (.NET) 中的响应式编程

C# (.NET) 中的响应式编程：Observable、Rx.NET 和数据流处理

响应式编程 (Reactive Programming) 是一种面向异步数据流处理和变化传播的编程范式。在 C# (.NET) 生态系统中，这一概念通过 Rx.NET (Reactive Extensions) 库以及 IObservable<T> / IObserver<T> 接口实现。与传统的命令式方法不同，响应式编程允许以声明式方式描述系统应如何响应事件，而不会阻塞主执行线程。

在 PythonLib 页面上，我们将探讨 C# 中响应式编程的关键组件，包括 Observable 序列、用于流转换的 LINQ 运算符以及实际使用示例。

C# 中响应式编程的核心概念

响应式编程基于 观察者 (Observer) 设计模式，其中：

Observable — 数据源（事件流）。
Observer — 订阅者，对传入的数据做出响应。
Subscription — 管理订阅的机制（从流中取消订阅）。

在 .NET 中，这些概念由 IObservable<T> 和 IObserver<T> 接口表示，但在实践中更常用的是 Rx.NET 库（NuGet 包 System.Reactive），它提供了丰富的运算符集用于处理序列。

安装 Rx.NET

首先，将 NuGet 包添加到项目中：

Install-Package System.Reactive

或者通过 .NET CLI：

dotnet add package System.Reactive

创建 Observable 并订阅数据流

考虑一个最简单的示例：创建一个 Observable，它延迟生成数字 1 到 5，然后订阅它。

using System;using System.Reactive.Linq;using System.Threading.Tasks;

class Program{    static async Task Main()    {        // 创建一个 Observable 序列        var observable = Observable.Interval(TimeSpan.FromSeconds(1))            .Select(x => x + 1) // 从 1 开始            .Take(5);           // 只取 5 个元素

        // 订阅流        var subscription = observable.Subscribe(            onNext: value => Console.WriteLine($"收到: {value}"),            onError: error => Console.WriteLine($"错误: {error.Message}"),            onCompleted: () => Console.WriteLine("流已完成")        );

        // 等待完成（在实际代码中使用 await 或阻塞）        await Task.Delay(6000);

        // 取消订阅（释放资源）        subscription.Dispose();    }}

用于流转换的 LINQ 运算符

Rx.NET 的主要优势之一是能够将 LINQ 中熟悉的运算符应用于数据流：Where、Select、Aggregate 等。这允许以声明式方式描述转换。

示例：过滤和转换

using System;using System.Reactive.Linq;

var numbers = Observable.Range(1, 10);

var processed = numbers    .Where(n => n % 2 == 0)    // 仅偶数    .Select(n => n * n)        // 求平方    .Do(n => Console.WriteLine($"已处理: {n}")); // 副作用

processed.Subscribe(    onNext: n => Console.WriteLine($"结果: {n}"),    onCompleted: () => Console.WriteLine("完成"));