Back

To Top

.NET 响应式编程 System.Reactive 系列文章（三）：Subscribe 和 IDisposable 的深入理解

2025/01/09

.NET 响应式编程 System.Reactive 系列文章（三）：Subscribe 和 IDisposable 的深入理解

在 Rx 中，Subscribe() 方法返回一个 IDisposable 接口对象，用于手动取消订阅和释放资源。另外，System.Reactive 还提供了不返回 IDisposable 的 Subscribe 重载，这些重载方法通过 CancellationToken 管理订阅的生命周期。在本篇文章中，我们将深入探讨 Subscribe 和 IDisposable 的原理、这些特殊重载的设计原因，以及在实际使用中的应用场景。

.NET 响应式编程 System.Reactive 系列文章（三）：Subscribe 和 IDisposable 的深入理解

在前两篇文章中，我们详细介绍了 IObservable<T> 和 IObserver<T> 的核心概念及交互流程。但在实际使用 System.Reactive 时，一个常见的误区是认为数据流一旦订阅，就不需要额外管理。这种认知是危险的，因为 Observable 的订阅可能是无限的，如果不管理好订阅的生命周期，很容易导致内存泄漏和资源浪费。

在 Rx 中，Subscribe() 方法返回一个 IDisposable 接口对象，用于手动取消订阅和释放资源。另外，System.Reactive 还提供了不返回 IDisposable 的 Subscribe 重载，这些重载方法通过 CancellationToken 管理订阅的生命周期。在本篇文章中，我们将深入探讨 Subscribe 和 IDisposable 的原理、这些特殊重载的设计原因，以及在实际使用中的应用场景。

Subscribe 是连接 IObservable<T> 和 IObserver<T> 的桥梁。当你调用 Subscribe() 方法时：

IObservable<T> 开始向 IObserver<T> 推送数据。
订阅会保持活跃状态，直到：
- 数据流结束（调用 OnCompleted()）。
- 发生错误（调用 OnError()）。
- 手动取消订阅（调用 Dispose()）。
- 超时取消订阅（向CancellationToken注册超时回调）。

普通的 Subscribe 重载 返回一个 IDisposable 对象，允许你通过调用 Dispose() 方法取消订阅。这是管理数据流生命周期的核心机制之一。

System.Reactive 提供了一些特殊的 Subscribe 重载方法，它们不返回 IDisposable，而是依赖于 CancellationToken 来控制订阅的生命周期。这些方法设计的目的是为了提供一种外部取消订阅的机制，让你无需手动管理 Dispose() 的调用。

2.1 方法签名

以下是其中一个不返回 IDisposable 的 Subscribe 重载：

public static void Subscribe<T>(
    this IObservable<T> source,
    Action<T> onNext,
    Action<Exception> onError,
    Action onCompleted,
    CancellationToken cancellationToken
);

这种重载方法的使用场景是：你希望通过 CancellationToken 来控制订阅的生命周期，而不是手动调用 Dispose()。

2.2 示例代码：使用 CancellationToken 管理订阅

示例：超时取消订阅

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

class Program
{
    static void Main(string[] args)
    {
        IObservable<long> observable = Observable.Interval(TimeSpan.FromSeconds(1));

        CancellationTokenSource cts = new();

        // 使用 Subscribe 方法并传入 CancellationToken
        observable.Subscribe(
            onNext: static value => Console.WriteLine($"Received: {value}"),
            onError: static ex => Console.WriteLine($"Error: {ex.Message}"),
            onCompleted: static () => Console.WriteLine("Completed"),
            token: cts.Token
        );

        // 模拟运行 5 秒后取消订阅
        Console.WriteLine("Running for 5 seconds...");
        Thread.Sleep(5000);
        cts.Cancel();
        Console.WriteLine("Subscription cancelled.");
    }
}

输出结果：

Running for 5 seconds...
Received: 0
Received: 1
Received: 2
Received: 3
Subscription cancelled.

2.3 使用场景：什么时候使用 CancellationToken？

使用场景

推荐的 Subscribe 重载

需要手动取消订阅

返回 IDisposable 的重载

使用外部控制（如用户交互、超时）控制订阅

带 CancellationToken 的重载

典型场景：

异步任务取消
在异步任务中使用 CancellationToken 取消订阅数据流，避免阻塞或内存泄漏。
超时控制
使用 CancellationTokenSource.CancelAfter() 设置超时取消订阅。

2.4 示例：设置超时取消订阅

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

class Program
{
    static void Main(string[] args)
    {
        IObservable<long> observable = Observable.Interval(TimeSpan.FromSeconds(1));

        CancellationTokenSource cts = new();
        cts.CancelAfter(TimeSpan.FromSeconds(3)); // 设置 3 秒后自动取消订阅

        observable.Subscribe(
            onNext: static value => Console.WriteLine($"Received: {value}"),
            onError: static ex => Console.WriteLine($"Error: {ex.Message}"),
            onCompleted: static () => Console.WriteLine("Completed"),
            token: cts.Token
        );

        Console.WriteLine("Running...");
        Thread.Sleep(5000);
        Console.WriteLine("Program ended.");
    }
}

输出结果：

Running...
Received: 0
Received: 1
Received: 2
Program ended.

3. 使用场景总结

使用方式

特点

适用场景

Subscribe 返回 IDisposable

允许手动取消订阅

长时间订阅或频繁管理多个订阅

Subscribe 接受 CancellationToken

通过外部控制（如超时或用户交互）取消订阅

异步任务、超时控制、用户交互场景

4. 注意事项：CancellationToken 的局限性

虽然使用 CancellationToken 可以简化订阅管理，但也有一些需要注意的地方：

不支持手动取消
如果你使用的是返回 IDisposable 的 Subscribe 方法，你可以手动调用 Dispose() 取消订阅。但如果你使用带 CancellationToken 的重载，就无法通过 Dispose() 取消订阅。
更适合一次性订阅
带 CancellationToken 的 Subscribe 重载更适合一次性订阅的场景。如果你需要频繁管理多个订阅，使用 CompositeDisposable 或手动管理 IDisposable 可能更合适。

5. 两种订阅方式的对比

特性

返回 IDisposable 的 Subscribe

带 CancellationToken 的 Subscribe

是否支持手动取消订阅

✅ 支持

❌ 不支持

是否支持外部控制订阅生命周期

❌ 需要手动调用 Dispose()

✅ 通过 CancellationToken 控制

是否适合长期订阅

✅ 适合

❌ 更适合一次性订阅

sequenceDiagram participant Observer as IObserver participant Observable as IObservable participant IDisposable as IDisposable Observer ->> Observable: Subscribe() Observable ->> Observer: OnNext(T value) Observable ->> Observer: OnNext(T value) Observer ->> IDisposable: Dispose() Observable -->> Observer: 停止推送数据

总结

在本篇文章中，我们详细探讨了 Subscribe 和 IDisposable 的内部机制，并重点介绍了 带 CancellationToken 的 Subscribe 重载：

Subscribe() 方法返回 IDisposable，用于管理订阅的生命周期。
不返回 IDisposable 的 Subscribe 重载，通过 CancellationToken 控制订阅的终止。
使用场景不同：IDisposable 更适合长期订阅，CancellationToken 更适合一次性或外部控制的订阅。

下一篇文章预告

《.NET 响应式编程 System.Reactive 系列文章（四）：操作符基础》
下一篇文章将介绍 System.Reactive 的基础操作符，包括如何创建、转换和过滤数据流。我们将通过实战示例，帮助你快速掌握 Rx 的操作符使用方法。敬请期待！

作者：VAllen
出处：http://www.cnblogs.com/vallen
本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文连接，否则保留追究法律责任的权利。
唯有偏执者得以生存。

转载声明
本文内容出自网络，非原创作品。由于无法确认原始来源和作者信息，在此对原作者表示感谢。
如涉及版权问题，请联系 [联系邮箱]，我们将及时处理。