【.NET跨平台】ReactiveUI实战：构建线程安全的动态数据集合

1. 为什么需要线程安全的动态数据集合在.NET开发中我们经常遇到需要在后台线程处理数据然后更新到UI界面的场景。传统的ObservableCollection虽然能通知UI更新但它有个致命缺陷——只能在创建它的线程通常是UI线程上操作。我曾在电商价格监控项目中踩过坑当多个爬虫线程同时更新商品列表时直接操作ObservableCollection会导致程序崩溃。举个例子假设我们要开发一个股票行情看板// 危险代码多线程操作ObservableCollection会导致崩溃 var stocks new ObservableCollectionStock(); Parallel.ForEach(stockCodes, code { var price FetchStockPrice(code); stocks.Add(new Stock(code, price)); // 这里会抛出跨线程异常 });ReactiveUI提供的SourceList和SourceCache就是为了解决这个问题而生的。它们内部实现了线程安全机制允许你在任何线程修改数据自动同步到UI线程。实测下来这种方案比手动调用Dispatcher.BeginInvoke要稳定得多代码也更简洁。2. ReactiveUI动态集合核心组件解析2.1 SourceList vs SourceCache的选择SourceList和SourceCache都是线程安全的动态集合但适用场景不同SourceList适合顺序集合类似加强版ListSourceCache适合需要键值查询的场景内部使用Dictionary维护我做过性能对比测试处理10万条数据操作类型SourceList耗时SourceCache耗时批量添加120ms150ms随机查询230ms50ms条件删除180ms90ms2.2 与ReadOnlyObservableCollection的配合ReactiveUI的黄金组合模式private readonly ReadOnlyObservableCollectionStock _visibleStocks; public ReadOnlyObservableCollectionStock VisibleStocks _visibleStocks; private readonly SourceCacheStock, string _allStocks new(x x.Code); // 初始化绑定 _allStocks.Connect() .Filter(x x.Price 0) // 动态过滤条件 .Sort(SortExpressionComparerStock.Descending(x x.Volume)) .Bind(out _visibleStocks) .Subscribe();这种模式有三大优势线程安全可以在任何线程调用_allStocks.AddOrUpdate(stock)自动同步数据变化会自动反映到_visibleStocks声明式编程通过Filter/Sort定义业务规则无需手动维护集合状态3. 实战构建实时日志监控系统最近用这套方案实现了一个分布式系统的日志看板核心代码如下3.1 ViewModel定义public class LogMonitorViewModel : ReactiveObject { private readonly SourceListLogEntry _logSource new(); private readonly ReadOnlyObservableCollectionLogEntry _logs; public ReadOnlyObservableCollectionLogEntry Logs _logs; [Reactive] public LogLevel MinimumLevel { get; set; } LogLevel.Info; public LogMonitorViewModel() { // 动态过滤自动排序 _logSource.Connect() .Filter(entry entry.Level MinimumLevel) .Sort(SortExpressionComparerLogEntry.Descending(x x.Timestamp)) .Bind(out _logs) .Subscribe(); } // 线程安全的添加日志方法 public void AddLog(LogEntry entry) _logSource.Add(entry); }3.2 多线程写入示例// 在任意线程调用都不会有问题 Parallel.For(0, 100, i { var log new LogEntry($Processing item {i}, LogLevel.Info); viewModel.AddLog(log); });3.3 性能优化技巧批量操作对于大批量数据使用Edit方法减少通知次数_logSource.Edit(innerList { foreach(var item in newItems) { innerList.Add(item); } });缓冲处理高频更新时使用BufferObservable.Interval(TimeSpan.FromSeconds(1)) .ObserveOn(TaskPoolScheduler.Default) .Buffer(TimeSpan.FromMilliseconds(500)) // 500ms缓冲窗口 .Subscribe(values { _logSource.AddRange(values.Select(x new LogEntry($Tick {x}))); });4. 高级应用场景与避坑指南4.1 与Blazor/WASM的集成在WebAssembly环境下由于单线程特性仍然需要注意// WASM需要特殊处理 _allStocks.Connect() .ObserveOn(RxApp.MainThreadScheduler) .Bind(out _visibleStocks) .Subscribe();4.2 内存泄漏预防动态集合常见的内存泄漏问题忘记处理订阅长期持有集合引用正确做法// 在ViewModel销毁时清理资源 this.WhenActivated(disposables { _allStocks.Connect() .Bind(out _visibleStocks) .Subscribe() .DisposeWith(disposables); });4.3 与Entity Framework配合在数据库查询场景中的典型用法var dbChanges _dbContext.Stocks .AsObservable() // 使用System.Linq.Async .Subscribe(stocks { _allStocks.Edit(inner { inner.Clear(); inner.AddRange(stocks); }); });