.NET(C#)中Reactive Extensions问题：Subject<T>仅处理一次订阅

你遇到的问题核心在于**Subject<T>在接收到OnCompleted通知后，会彻底停止处理后续的所有事件**。

咱们来拆解下你的代码执行流程，就能明白为啥第二个请求没反应：

1. 你创建了Subject<string>实例messenger，并订阅了它的过滤流。
2. 第一个Observable.Return("File 1")会先调用messenger.OnNext("File 1")，紧接着自动调用messenger.OnCompleted()——这是Observable.Return的默认行为：发射单值后立即标记流完成。
3. 当Subject收到OnCompleted通知后，会自动取消所有已订阅的观察者，并且拒绝接收后续任何OnNext或OnError通知。这就是第二个Observable.Return("File 2")调用OnNext时毫无反应，且HasObservers变为false的原因。

可选解决方案 #

根据你的实际场景，有几种修复方式：

1. 使用支持持续接收事件的Subject变体 #

如果你的场景需要持续处理外部请求，不要让Subject被完成，可以改用以下类型：

• BehaviorSubject<T>：会保留最后一次发射的值，新订阅者会立即收到这个值，后续的新事件也能正常接收。
• ReplaySubject<T>：可以缓存指定数量的历史事件，新订阅者会收到缓存的事件，同时也会接收后续的新事件。

修改后的示例代码：

using System.Reactive.Linq;
using System.Reactive.Subjects;

static void Main(string[] args) { 
    // 替换为BehaviorSubject，初始化值可设为null或空字符串
    BehaviorSubject<string> messenger = new BehaviorSubject<string>(string.Empty); 
    messenger.Where(o => o.Length > 0).Subscribe(file => { 
        Console.WriteLine("got file request: " + file); 
    }); 
    var pathObservable = Observable.Return<string>("File 1"); 
    pathObservable.Subscribe(messenger); 
    var pathObservable2 = Observable.Return<string>("File 2"); 
    pathObservable2.Subscribe(messenger); 
    Console.ReadKey(); 
}

2. 阻止Observable.Return传递OnCompleted通知 #

如果你必须使用普通Subject<T>，可以在订阅时只传递OnNext和OnError，忽略OnCompleted：

// 替换原来的订阅写法
pathObservable.Subscribe(
    onNext: messenger.OnNext,
    onError: messenger.OnError
    // 不传递onCompleted参数，或者显式设为null
);

这样Subject就不会收到完成通知，就能持续处理后续的事件了。

3. 用Publish/RefCount创建持久化共享流 #

如果你的外部事件流需要被多个观察者共享，也可以用Publish()和RefCount()创建一个不会自动完成的共享流，不过这个方案更适合多观察者共享同一数据源的场景。

总结 #

你之前的核心误解是忽略了Observable.Return的默认行为——它发射单值后会立即触发OnCompleted，而普通Subject在完成后就彻底停止工作了。选择适合你业务场景的Subject变体，或者控制订阅时的通知传递，就能解决这个问题。

内容的提问来源于stack exchange，提问作者Greg