为何我的Rx.NET可观察序列会完整输出两次？

看到你遇到的随机失败问题——预期只收到一次[8,10,11]，但偶尔会收到两次重复的序列，这在Rx.NET的测试里通常和冷序列的订阅行为或者扩展方法/测试代码中的重复订阅逻辑有关，我来帮你拆解可能的原因和解决办法：

一、先排查最常见的原因：冷序列被多次订阅 #

Rx.NET里的大多数基础Observable（比如Observable.Range、Observable.FromEnumerable）都是冷序列——每次有新订阅时，都会从头执行一遍序列的生成逻辑。如果你的测试或扩展方法中不小心对同一个冷序列发起了两次订阅，就会得到两次完整的输出。

排查点： #

• 检查你的扩展方法实现：有没有在内部多次订阅源序列？比如误写了source.Concat(source)、SelectMany(_ => source)这类会重复引入源序列的逻辑？
• 检查测试代码：是不是在测试流程中对目标Observable订阅了两次？比如先用了Subscribe()，又调用了ToListAsync().Wait()，或者测试框架的断言逻辑隐性触发了第二次订阅？

解决办法： #

如果确认是重复订阅导致的，用多播操作符把冷序列转为热序列，确保所有订阅共享同一次发射：

// 用Publish+RefCount实现“当有订阅时激活，无订阅时销毁”的共享序列
var sharedSequence = source.MyExtension().Publish().RefCount();
// 后续多次订阅sharedSequence只会触发一次源序列发射
var result = await sharedSequence.ToListAsync();

二、检查扩展方法的逻辑是否存在重复发射 #

有时候扩展方法里的逻辑会无意中重复发射数据，比如：

• 错误地使用了Merge或Concat合并了同一个序列两次
• 在Select或Where之外额外添加了重复的发射逻辑
• 误用了Repeat操作符（哪怕是无意的）

举个错误示例：

// 错误：Concat了同一个源序列两次，导致输出重复
public static IObservable<int> MyFilterExtension(this IObservable<int> source)
{
    var filtered = source.Where(x => x > 5);
    return filtered.Concat(filtered); // 这里会导致序列重复
}

排查方法： #

把扩展方法的逻辑简化，先去掉非核心转换，测试最基础的源序列输出，逐步添加逻辑定位问题点；或者用Rx.NET的Do操作符打印每个步骤的发射记录：

source.Do(x => Console.WriteLine($"Source emitted: {x}"))
      .MyExtension()
      .Do(x => Console.WriteLine($"Extension emitted: {x}"))
      .ToList()
      .Wait();

通过日志就能看到是不是源序列被发射了两次，还是扩展方法内部重复发射了。

三、用TestScheduler消除调度器的随机性 #

随机失败很大概率和调度器的不确定性有关——比如你在测试中用了默认的CurrentThreadScheduler，异步操作的订阅时机可能受线程调度影响，偶尔导致重复订阅。

改用TestScheduler来做单元测试，能精确控制时间和订阅行为，彻底消除随机性：

using Microsoft.Reactive.Testing;

[Test]
public void MyExtension_ShouldEmitExpectedValuesOnce()
{
    var scheduler = new TestScheduler();
    
    // 创建一个热序列，提前定义好发射的时间和数据
    var source = scheduler.CreateHotObservable(
        ReactiveTest.OnNext(100, 8),
        ReactiveTest.OnNext(200, 10),
        ReactiveTest.OnNext(300, 11),
        ReactiveTest.OnCompleted<int>(400)
    );

    // 启动测试，精确控制创建、订阅、销毁的时间点
    var result = scheduler.Start(() => source.MyExtension(),
                                created: 0,
                                subscribed: 50,
                                disposed: 500);

    // 断言结果
    var expectedValues = new[] { 8, 10, 11 };
    var actualValues = result.Messages.Select(msg => msg.Value.Value);
    CollectionAssert.AreEqual(expectedValues, actualValues);
}

用TestScheduler能让测试完全可复现，再也不会出现随机失败的情况。

四、其他可能的小坑 #

• 检查是不是测试类的Setup方法重复创建了Observable实例，导致每次测试都叠加了订阅？
• 如果你用了ReplaySubject或BehaviorSubject，是不是之前的测试残留了数据？记得在每次测试后清理Subject的状态。

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