企业官网建设流程全解析

先从结论说起：标题里的mono subscribe，我按最实际的项目语境把它拆成两层——mono指 Mono / Unity 这类跨平台 C# 运行环境，subscribe指System.Reactive（Rx）里那条让所有数据流真正动起来的调用链。这篇源码分析不打算把 Rx 的全部算子都过一遍，只想把Subscribe从扩展方法入口到冷流工厂、热流注册表、可释放链、调度器这几层逐个掰开，最后落到 Mono 平台上最容易踩的三个坑上。适合正在用 Rx 写跨平台服务、Unity 客户端，或者刚接触响应式编程想把底层搞清楚的开发者。

为什么执着于这条链？因为项目里十有八九的疑难问题都出在“订阅没生效 / 订阅泄漏 / 回调跑了别的线程”上，而这些现象的背后全部指向Subscribe的实现细节。把它读透，后面看什么算子都顺。

1. 先定位：mono 和 subscribe 这一次到底碰在哪一层

1.1 标题里三个关键词的歧义

中文技术圈说“mono”，至少有三层意思：Mono 运行时、Unity 里常用的 UniRx、以及System.Reactive里的标准实现。很多人第一次搜这句话，其实是想搞清楚“我在 Unity 里写下.Subscribe(...)，这行代码背后干了什么”。

这三者的内核都是观察者模式，但订阅语义、异常策略、调度行为差异很大。本文以System.Reactive的源码为解剖台，UniRx的差别会放在相关章节单独说。

1.2 协议这东西：IObserver 的三个回调不是随便实现的

先看接口本身：

public interface IObserver<in T> { void OnNext(T value); void OnError(Exception error); void OnCompleted(); } public interface IObservable<out T> { IDisposable Subscribe(IObserver<T> observer); }

光看签名容易忽略一个东西：IObserver是一份“非强制协议”。所谓非强制，是指编译器不会检查调用顺序，但所有 Rx 算子都默认你会遵守这三条规则：

• OnNext可以被调用零次或多次；
• OnError和OnCompleted二选一，且最多一次；
• 进入终止状态后，不允许再调用OnNext。

在项目里我见过有人为了实现一个简单监听器，在OnError之后继续OnNext，结果后续算子的内部状态彻底错乱，表现出来的现象是数据偶尔重复、偶尔丢失。这跟Subscribe源码没有直接关系，但理解整个链路的起点就是这份协议。

1.3 冷流与热流：Subscribe 触发的是“开始”而不是“连接”

很多初学者把Subscribe理解成“把线接上，以后一直收数据”，这是错的。准确地说，Subscribe是“发起一次执行”。这个区别在冷流和热流上尤其明显：

• 冷流：每次Subscribe都会重新执行一遍源头逻辑。Observable.Create、Observable.Range都属于这类，订阅一次就重新生产一轮数据。
• 热流：Subject这类对象本身就在广播，Subscribe做的事情更像是在广播名单里登记一个新接收者，后续数据继续推给它。

后面第 3、4 章会分别从冷流和热流的源码来看Subscribe的实现差异。理解这两条分支，才算真正跨进了 Rx 的门。

2. 入口层读起：ObservableExtensions.Subscribe 是怎么把回调包成 Observer 的

2.1 最省事的重载：不带参数的 Subscribe

大多数业务代码里写的都是Subscribe(x => ...)这种带回调的形式，但源码最先提供的其实是一个不带任何参数的重载：

public static IDisposable Subscribe<T>(this IObservable<T> source) { if (source == null) throw new ArgumentNullException(nameof(source)); return source.Subscribe(new DefaultObserver<T>()); }

DefaultObserver的实现很朴素：

internal sealed class DefaultObserver<T> : IObserver<T> { public void OnNext(T value) { } public void OnError(Exception error) { throw error; } public void OnCompleted() { } }

注意OnError是直接throw。如果上游发生了异常而这个重载没有回调来兜底，异常会被抛到当前上下文之外。在同步流里它会直接炸出调用栈，在异步流里则可能落到线程池的未处理异常处理器上，最终表现为“程序没报错但线程池在刷日志”。

这才是Subscribe()无参数重载最容易被误解的地方：看起来什么都不处理，实际上异常默认往外抛，而且抛的位置很难预料。

2.2 三个回调的适配：Observer.Create 与 AnonymousObserver

带一个onNext回调的版本是这样的：

public static IDisposable Subscribe<T>(this IObservable<T> source, Action<T> onNext) { if (source == null) throw new ArgumentNullException(nameof(source)); if (onNext == null) throw new ArgumentNullException(nameof(onNext)); return source.Subscribe(Observer.Create<T>(onNext)); }

真正被Subscribe接收的是一个IObserver<T>，而不是Action<T>。所以库内部需要把回调包装成观察者，这个包装类型叫AnonymousObserver：

internal sealed class AnonymousObserver<T> : ObserverBase<T> { private readonly Action<T> _onNext; private readonly Action<Exception> _onError; private readonly Action _onCompleted; public AnonymousObserver(Action<T> onNext, Action<Exception> onError, Action onCompleted) { _onNext = onNext ?? throw new ArgumentNullException(nameof(onNext)); _onError = onError ?? throw new ArgumentNullException(nameof(onError)); _onCompleted = onCompleted ?? throw new ArgumentNullException(nameof(onCompleted)); } protected override void OnNextCore(T value) => _onNext(value); protected override void OnErrorCore(Exception error) => _onError(error); protected override void OnCompletedCore() => _onCompleted(); }

也就是说，Subscribe(onNext: x => ...)最终会变成source.Subscribe(new AnonymousObserver<T>(x => ..., Stubs.Throw, Stubs.Nop))。这里Stubs.Throw是默认的异常处理函数，作用是把异常原样抛出；Stubs.Nop是空的完成回调。

有一个我在项目里踩过的点：既然Observer.Create会把缺失的onError补成Stubs.Throw，那么你只传onNext时以为“没报错就是安全”，其实异常根本没有被捕获，只是被丢到了别处。排查这类问题时，重点要看线程池日志和全局异常钩子。

2.3 参数缺省时那些容易误读的“默认行为”

不同版本的 Rx 对缺省参数的处理有细微差异，这里以常见版本为例列一下：

写代码的人经常只关心onNext，忽略了onError和onCompleted。实际上在异步数据源里，onCompleted往往是你唯一能安全做资源清理的地方，不传这个回调，有些清理点就得靠别的机制补。

2.4 扩展方法入口断点技巧

因为Subscribe是扩展方法，调试的时候断点下在业务代码的.Subscribe那一行往往只能看到调用发生，看不到内部状态。我的习惯是在Observer.Create和source.Subscribe(observer)这两处下断点。前一个能确认回调是否被正确包装，后一个能确认调用是否真正进入了 Observable 实现。对于复杂调用链，再配合“调用堆栈”窗口从底往上数，基本能定位是哪一层算子改变了行为。

3. 冷流侧：Observable.Create 里那个工厂方法，才是主体

3.1 AnonymousObservable 包装了什么

Observable.Create是冷流体系里最常用的入口，它的实现非常短：

public static IObservable<T> Create<T>(Func<IObserver<T>, IDisposable> subscribe) { if (subscribe == null) throw new ArgumentNullException(nameof(subscribe)); return new AnonymousObservable<T>(subscribe); }

内部类型：

internal sealed class AnonymousObservable<T> : ObservableBase<T> { private readonly Func<IObserver<T>, IDisposable> _subscribe; public AnonymousObservable(Func<IObserver<T>, IDisposable> subscribe) { _subscribe = subscribe; } protected override IDisposable SubscribeCore(IObserver<T> observer) { return _subscribe(observer) ?? Disposable.Empty; } }

关键点在于“延迟执行”。Observable.Create只是把你的工厂函数存起来，一等公民，不会立刻调用。真正调用它的时机在Subscribe发生的那一刻。冷流的整个生命周期，其实是从Subscribe的括号开始，到工厂函数执行完成为止。

我把Observable.Create理解成一个“生产计划单”，Subscribe是“开工指令”。你不开工，计划单永远是计划单。

3.2 ObserverBase 对协议顺序的强制

AnonymousObservable继承ObservableBase<T>，ObservableBase里有一套对协议顺序的保护逻辑。核心代码如下：

public abstract class ObserverBase<T> : IObserver<T>, IDisposable { private int _isStopped; public void OnNext(T value) { if (Volatile.Read(ref _isStopped) == 0) { try { OnNextCore(value); } catch { Dispose(); throw; } } } public void OnError(Exception error) { ... } public void OnCompleted() { ... } protected abstract void OnNextCore(T value); protected abstract void OnErrorCore(Exception error); protected abstract void OnCompletedCore(); }

_isStopped一旦置位，后续OnNext会被直接忽略。这对异常安全很重要：回调抛异常时，订阅会被自动 Dispose，防止同一批数据反复触发同一段出错逻辑。上游工厂如果违反协议在OnCompleted后继续OnNext，下游的观察者会自动把消息吞掉。

这里有一个项目里容易踩的坑：上游工厂自己写得很随意，比如先调了observer.OnCompleted()又调observer.OnNext(value)，你以为消息发出去了，实际却被ObserverBase拦截。调试时如果发现“数据没到”，先查工厂内部的调用顺序，而不是怀疑下游逻辑。

3.3 依赖收尾：factory return 的 IDisposable 何时触发

冷流工厂返回值是IDisposable，这个返回值在ObservableBase.Subscribe内部会被挂到“订阅”这个整体对象上。同步流里，工厂如果调用了observer.OnNext之后马上observer.OnCompleted，那么终止信号会触发依赖释放，工厂返回的IDisposable.Dispose()会立即被执行。

看一个实际例子：

var seq = Observable.Create<int>(observer => { Console.WriteLine("factory start"); observer.OnNext(1); observer.OnNext(2); observer.OnCompleted(); Console.WriteLine("factory end"); return Disposable.Create(() => Console.WriteLine("cleanup")); }); seq.Subscribe( x => Console.WriteLine(x), ex => Console.WriteLine(ex), () => Console.WriteLine("done"));

执行顺序是：

factory start 1 2 done factory end cleanup

这里有一个容易误判的顺序：cleanup的打印在factory end之后。原因在于OnCompleted只是发出终止信号，真正执行Dispose要等工厂函数返回、把IDisposable交给订阅框架之后，框架才能调用它。所以如果你的工厂函数里有耗时的尾操作，清理动作会被推迟——这在异步流里尤其明显。

4. 热流侧：Subject.Subscribe 里的注册表与退订

4.1 subscribe 就是数组 add

冷流每次Subscribe都“重新生产”，热流则完全不同。以最常见的Subject<T>为例，它的内部维护着一个观察者数组：

public IDisposable Subscribe(IObserver<T> observer) { if (observer == null) throw new ArgumentNullException(nameof(observer)); lock (_gate) { if (_isDisposed) throw new ObjectDisposedException(nameof(Subject<T>)); if (_isStopped) { if (_error != null) observer.OnError(_error); else observer.OnCompleted(); return Disposable.Empty; } _observers = _observers.Add(observer); return new SubjectSubscription(this, observer); } }

这段代码本质上是：在锁里把observer追加进数组，然后返回一个“退订凭证”。所谓退订凭证，就是SubjectSubscription，它的Dispose会把observer从数组里移除。

很多人问，Subject 和普通事件有什么区别？区别就在于这个数组的公开性。事件用+=注册，Subject 用Subscribe注册，前者没有返回值，后者返回IDisposable。这个返回值的存在，意味着退订是一个一等公民操作，可以被算子自由传递和组合。

4.2 终止状态下的后退路：OnError 和 OnCompleted

注意上面代码里那段：如果 Subject 已经处于终止状态，新来的订阅者不会进入数组，而是立刻收到OnError或OnCompleted。这是热流和冷流又一个大差别。

冷流的终止只影响当前这次订阅；热流终止之后再来订阅的人，会直接收到“历史终态”。这在业务上很有用：一个全局状态流已经完成，之后新模块接入时不需要等下一次事件，直接拿到终态即可。

但这也带来一个坑：如果你写的 Subject 是全局单例，且在某次错误后进入了终止状态，那么所有后续订阅都会立刻收到同一个错误。想让它继续工作，只能重新创建 Subject，不能用旧的。这个行为在源码里看得一清二楚。

4.3 真退订时锁与数组置换

SubjectSubscription的退订实现如下（代码做了简化，保留主干）：

private sealed class SubjectSubscription : IDisposable { private readonly Subject<T> _subject; private IObserver<T> _observer; public void Dispose() { IObserver<T> current = _observer; if (current != null) { _observer = null; _subject.Unsubscribe(current); } } } private void Unsubscribe(IObserver<T> observer) { lock (_gate) { _observers = _observers.Remove(observer); } }

退订不是把数组某个位置置空，而是用“移除后重新生成数组”的方式，返回一个新的引用。这样做是为了让OnNext在遍历时可以安全地使用快照，避免遍历过程中对方忽然被释放导致空引用。代价是频繁订阅退订时会有数组拷贝开销。

在项目里滥用Subscribe却不退订，Subject 的数组就会不断变大。这个增长不像普通事件那样直接表现在内存暴涨上，而是像一个隐藏的慢泄漏，到达一定规模后才开始拖垮OnNext的遍历速度。

5. 订阅的身份证：一串麻烦的 IDisposable

5.1 Disposable.Create 的原子性

Rx 几乎把所有资源清理都统一成IDisposable，静态工厂Disposable.Create就是最常见的入口：

public static IDisposable Create(Action disposeAction) { if (disposeAction == null) throw new ArgumentNullException(nameof(disposeAction)); return new AnonymousDisposable(disposeAction); }

AnonymousDisposable内部用了原子操作，保证Dispose只执行一次。这是响应式编程里面特别容易忽略的细节：上游和下游可能会同时触发清理，如果Dispose不是幂等的，就会重复释放同一条连接、重复关闭同一个资源。

实际项目里经常见到有人自己写：

return Disposable.Create(() => { client.Dispose(); channel.Close(); });

如果client.Dispose()本身不幂等，而这条订阅被Take(1)和TakeUntil同时挂上，清理逻辑可能被执行两次。轻则日志噪音，重则直接异常。所以所有自定义清理回调，都应该按“可能执行多次但仍然安全”的标准写。

5.2 SingleAssignmentDisposable 与 SerialDisposable 的分工

IDisposable不只是给用户用的，Rx 内部算子也在用，而且用的是更细的容器类：

SerialDisposable是最能体现“切换”语义的。当流从 A 切换到 B 时，它会把对 A 的订阅先 Dispose 掉，再挂上对 B 的订阅。如果你自己实现一个带切换功能的数据源，直接用SerialDisposable比自己管理旧订阅要安全得多。

5.3 CompositeDisposable：页面级“总开关”

CompositeDisposable在客户端项目里几乎是标配。一屏数据可能包含网络请求、按钮点击、定时器等多条订阅，把它们全部装进一个CompositeDisposable，页面销毁时一次性释放：

private readonly CompositeDisposable _cleanup = new(); void StartListening(IObservable<int> source) { source .Subscribe(x => Handle(x)) .AddTo(_cleanup); } void OnDestroy() { _cleanup.Dispose(); }

这里.AddTo(_cleanup)做的事情，本质上就是“把订阅返回的 IDisposable 塞进一个总容器”。总容器销毁时逐一释放，等价于对每一条订阅调用退订。在 Unity 里还有一种变体，直接把生命周期对象传给扩展方法，让容器跟随场景对象自动销毁，原理相同。

项目里排查泄漏时，第一件事就是检查每个Subscribe的返回值有没有被收集。只要有任何一个没收集，而这又恰好在热流上，泄漏就成立了。

6. 调度器掺和之后，Subscribe 返回的瞬间不代表推完了

6.1 同步工厂：订阅线程即执行线程

冷流工厂如果不主动开线程，数据流是同步的：Subscribe调用在哪条线程，OnNext就在哪条线程执行。这一点简单直接，但经常被误解。

不少项目里会遇到一个奇怪现象：Subscribe之后的下一行代码，竟然先于OnNext执行。这件事得看工厂是否真的同步。如果工厂里面用了async、Task.Run或者调度器，那数据流的执行可能被丢到别的线程，Subscribe自然就提前返回了。

换句话说，“同步 / 异步”不由Subscribe这个入口决定，而是由整条链路上的调度器决定。这也是为什么源码分析不能只看入口，还要看调度器。

6.2 ObserveOn 是把回调挪到另一条线程的关卡

ObserveOn是调度器介入最典型的地方。它的底层实现是把上游消息先投递到一个调度器队列，再由调度器在目标线程上逐个执行OnNext：

public static IObservable<T> ObserveOn<T>(this IObservable<T> source, IScheduler scheduler) { if (source == null) throw new ArgumentNullException(nameof(source)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return new ObserveOnObservable<T>(source, scheduler); }

ObserveOnObservable.SubscribeCore里会创建一个ObserveOn<T>.Sink，Sink负责把上游的OnNext转成调度器上的工作项。整个过程可以简单理解成：上游线程把车开到调度器门口，调度器再按自己的规则决定什么时候、在哪条线程继续开。

所以出现“回调不在我 Subscribe 的线程”时，优先查链路上有没有ObserveOn、SubscribeOn、DelaySubscription这类算子，而不是怀疑操作系统线程调度。

6.3 Unity / Mono 主线程调度的现实折腾

Mono 环境里最典型的调度问题是 Unity。Unity API 大部分要求在主线程调用，而网络回调天然落在其他线程。如果直接让异步回调更新 UI，会报类似“Unity API can only be called from main thread”的异常。

做法是把跨线程流切成两段：

networkObservable .ObserveOn(mainThreadScheduler) .Subscribe(x => UpdateUI(x));

mainThreadScheduler需要自己实现，或者用 UniRx 自带的ObserveOnMainThread()。原理并不复杂：让一个主线程的更新循环不断从队列里取消息，把OnNext挪到主线程执行。理解了ObserveOn的源码结构，就能看懂这个调度器只是换了一个IScheduler实现而已。

在 Mono 平台上，主线程调度器还要注意优先级和吞并问题：如果主线程本身在跑一个耗时循环，调度的消息会排队，回调延迟会变大。这不是源码层面的 bug，而是调度器的自然行为。

7. Mono 平台下的 AOT、内联与内存泄漏排查

7.1 断点进不去：内联和 JIT 的捣乱

Mono 的 JIT 会在发布构建里做大量内联，尤其AnonymousObservable.SubscribeCore这种只有一个函数调用的方法，很容易被内联掉。于是你下断点，会发现断点命中不了，或者跳到了奇怪的外层函数。

解决办法是在调试配置里关闭优化，或者给关键方法加[MethodImpl(MethodImplOptions.NoInlining)]。这个属性可以临时加上去，定位完问题再摘掉。还有另一个方法：不要给方法下断点，而是给工厂函数里的副作用语句下断点，比如给Console.WriteLine、Debug.Log下断点，因为副作用调用不会被内联到无可追踪的地步。

7.2 AOT 裁剪与泛型委托：真机才出的问题

Mono 这边有一个让很多人都头疼的场景：编辑器里跑得好好的，发布到手机 AOT 环境后就崩，而且崩的位置指向System.Reactive内部的泛型委托。

原因通常是两件事叠加：AOT 裁剪把某些反射入口裁掉了，而 Rx 的部分操作符在内部用表达式树编译生成委托。裁剪后的全 AOT 环境不支持动态编译，就会在调用入口处直接失败。严格来说这不是Subscribe本身的逻辑错误，而是平台约束。

应对方式有三条：

• 发布前就锁定目标机型和 AOT 配置，不要等用户报错；
• 用link.xml保留System.Reactive相关程序集；
• 避免在订阅回调里依赖反射动态生成代码，比如不要用Activator.CreateInstance去构造观察者，尽量用强类型写法。

在 Xamarin / Unity 项目里，这条属于“不踩不知道，踩了全是泪”的隐藏项。

7.3 泄漏排查路径：看的是持有链而不是引用的代码

项目里出现“Hot 流导致对象无法回收”的典型持有链是这样的：页面对象持有Subject或某个热流，而热流内部数组持有AnonymousObserver，AnonymousObserver又持有页面里的回调委托，委托捕获了页面对象。最终整个页面对象被自己创建的流持有，GC 无法回收。

排查时不要只看“谁引用了谁”，要打开内存分析工具，从 GC Root 出发看持有链。常见路径是：

线程 / 常驻对象 -> Subject<T>（热流广播器） -> observer 数组 -> AnonymousObserver<T> -> Action<T> 委托 -> 页面对象 / 大数组

定位到持有链后，按引用关系回溯，找到没有退订的那条Subscribe，把它收进CompositeDisposable，或者改成用Take(1)等自动终止算子。处理完再拍一次内存快照，看对象数量是否下降。

这个步骤说起来简单，但实际项目里往往有四五层引用包装，全靠看代码很难找全。所以我的建议是：一开始就给Subscribe的返回值落实“收集”纪律，而不是等到泄漏了再回头补。

最后说一点个人体会：读Subscribe源码，最有用的不是记住那些类名，而是记住三件事。第一，IObserver协议是运行时约定，顺序错、终态错都会引发诡异问题；第二，Observable.Create是延迟执行，真正的开始永远在Subscribe；第三，订阅必须持有返回的IDisposable，否则热流上必然泄漏。把这三件事刻在脑子里，再去读任何 Rx 算子，都会顺畅很多。