C# (.NET) 中的响应式编程：Rx.NET 与 IObservable

C# (.NET) 中的响应式编程：Rx.NET 完全指南

响应式编程是一种面向异步数据流和变化传播的开发范式。在 C# (.NET) 生态系统中，实现这一方法的主要工具是 Reactive Extensions (Rx.NET) 库。它允许您像通过 LINQ 处理普通集合一样，方便地处理随时间变化的数据序列。

在本文中，我们将解析响应式编程的关键概念，介绍 Rx.NET 的主要组件，并为教育网站 PythonLib 提供实用的代码示例。

响应式编程的核心概念

Rx.NET 的基础是两个接口：

• IObservable<T> — 数据源（可观察序列），可以发送元素、错误或完成信号。
• IObserver<T> — 订阅者（观察者），接收来自数据源的通知：OnNext、OnError、OnCompleted。

核心思想：订阅者不是轮询数据源（拉模型），而是对事件做出反应（推模型）。这对于处理 UI 事件、网络请求、传感器数据流或任何其他异步操作特别有用。

Rx.NET 入门

要在 C# 项目中使用 Rx.NET，请安装 NuGet 包：

dotnet add package System.Reactive

示例 1：创建并订阅 observable

让我们创建一个简单的数字序列，它延迟生成值：

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

class Program{    static async Task Main()    {        var observable = Observable            .Interval(TimeSpan.FromSeconds(1)) // 每秒生成 0, 1, 2...            .Take(5); // 只取前 5 个数字

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

        // 等待流完成        await Task.Delay(6000);        subscription.Dispose(); // 取消订阅    }}

示例 2：使用 LINQ 进行过滤和转换

Rx.NET 支持所有标准 LINQ 运算符，使代码更加简洁：

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

var processed = values    .Where(x => x % 2 == 0) // 仅偶数    .Select(x => x * 10)    // 乘以 10    .Timestamp();           // 添加时间戳

processed.Subscribe(x => Console.WriteLine($"[{x.Timestamp}] 值: {x.Value}"));

Rx.NET 的高级功能

合并流

您可以组合多个 observable 序列：

var stream1 = Observable.Interval(TimeSpan.FromSeconds(1)).Select(x => $"A{x}");var stream2 = Observable.Interval(TimeSpan.FromSeconds(1.5)).Select(x => $"B{x}");

var merged = stream1.Merge(stream2); // 合并为一个流merged.Take(6).Subscribe(Console.WriteLine);

多线程管理

Rx.NET 可以轻松切换执行上下文：

var uiScheduler = new SynchronizationContextScheduler(SynchronizationContext.Current);

Observable    .Interval(TimeSpan.FromSeconds(1))    .ObserveOn(uiScheduler) // 确保在 UI 线程中执行    .Subscribe(x => UpdateUI(x)); // 更新界面

错误处理

Rx.NET 提供了强大的故障恢复机制：

var faultyStream = Observable.Throw<int>(new Exception("错误!"));

var safeStr