Rx.NET按标的分组下每组独立BehaviorSubject结合Switch的实现与性能咨询

现有ReactiveSymbolFeed实现的问题与规模风险 #

• 引入的额外副作用
  • 生命周期管控缺失：构造函数内隐式触发订阅、没有暴露IDisposable释放接口，一旦应用退出或Symbol长期闲置，对应的Subject、Buffer窗口、订阅逻辑无法被主动清理，长期运行会出现内存泄漏。
  • 并发安全隐患：使用非线程安全的普通Dictionary做分发映射，当行情推送、阈值更新分属不同线程时，TryGetValue、Add操作存在竞态风险，容易触发运行时崩溃。
  • 逻辑耦合：阈值检测、输出动作（Console.WriteLine）被硬编码在包装类的构造函数中，无法灵活替换输出逻辑（比如对接消息队列、数据库），违背了Rx流可组合的核心特性。
• 大规模下的运行风险
  在每秒10万条消息、2000个分组的量级下，基础转发性能可以达标，但存在两个不可忽视的问题：
  • 没有背压隔离：手动foreach循环是同步阻塞处理逻辑，若某条分组的下游处理延迟升高（比如输出IO卡顿），会直接拖慢整个行情流的处理节奏，无超时、隔离机制。
  • 资源无清理机制：动态新增/删除Symbol时，手动维护的字典没有对应的回收逻辑，长期运行下闲置分组的资源会持续累积，最终引发内存溢出。

纯Rx算子链路的实现方案 #

完全不需要依赖手动包装类，通过原生Rx算子就可以实现“按Symbol分组、每个分组绑定独立阈值源、Switch切换最新配置”的需求，核心思路是利用GroupBy的自动分组能力，结合全局BehaviorSubject承载的阈值配置流，通过Switch完成阈值版本的自动切换。

核心实现代码 #

// 全局原始行情流（对接上游行情推送）
var ticks = new Subject<Tick>();
// 全局阈值配置流：BehaviorSubject承载每个Symbol的最新阈值，自动向订阅者推送最新配置
var thresholdConfig = new BehaviorSubject<IDictionary<string, double[]>>(
    new Dictionary<string, double[]>
    {
        ["AAPL"] = new[] {120.0, 125.0, 130.0},
        ["VXX"] = new[] {10.5, 15.0, 18.5, 20.0}
    });

var thresholdCrossedEventStream = ticks
    // 按Symbol自动分组，每个分组是独立的Tick流，由Rx框架管控生命周期
    .GroupBy(tick => tick.Symbol)
    .SelectMany(group =>
    {
        // 取出当前Symbol对应的最新阈值流，相同阈值不重复触发
        var currentSymbolThresholds = thresholdConfig
            .Select(config => config.TryGetValue(group.Key, out var levels) 
                ? levels 
                : Array.Empty<double>())
            .DistinctUntilChanged();

        // 配合Switch：始终使用最新的阈值配置执行检测，旧版本的检测逻辑自动终止
        return currentSymbolThresholds
            .Select(thresholds => group
                .Select(tick => (double)tick.Value)
                // 连续两个Tick窗口做跨阈值检测，与原单流逻辑完全一致
                .Buffer(2, 1)
                .Where(window => window.Count == 2)
                .Select(window => new
                {
                    CrossedLevels = thresholds.GetCrossovers(window[0], window[1]),
                    Previous = window[0],
                    Current = window[1]
                })
                .Where(x => x.CrossedLevels.Any())
                .SelectMany(x => x.CrossedLevels
                    .Select(level => new ThresholdCrossedEvent(level, x.Previous, x.Current)))
            )
            .Switch()
            // 可选：给长期无行情的分组加超时回收，避免闲置资源占用
            .TakeUntil(group.Throttle(TimeSpan.FromMinutes(10)));
    });

// 订阅输出流，可灵活替换后续处理逻辑
var eventSubscription = thresholdCrossedEventStream
    .Subscribe(x => Console.WriteLine(JsonConvert.SerializeObject(x)));

方案优势 #

• 无手动副作用：所有分组流、阈值流的生命周期由Rx框架统一管控，流结束、超时闲置时自动释放资源，不存在内存泄漏风险。
• 并发安全：Rx框架内置的调度器会保证流处理的顺序性，多线程推送行情、更新阈值时不存在竞态问题，无需额外加锁。
• 性能匹配规模：GroupBy的分组效率与手动Dictionary查找处于同一量级，2000个分组、每秒10万条消息的场景下，通过合理配置Scheduler可以利用多核并行处理，吞吐能力优于手动循环分发的实现。

结论 #

• 现有包装类实现确实引入了额外的生命周期、并发安全问题，在长时间运行的大规模行情场景下存在内存泄漏、延迟拖垮的风险。
• 纯Rx算子链路完全可以满足需求，且可维护性、安全性、性能都优于手动包装类实现。只有当需要对非Rx的 legacy 代码暴露命令式接口时，手动包装类才有存在的价值。

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