AvaloniaUI 中 Observable 与 ObserveOn 的用法和区别

1. Observable 基础概念

在 AvaloniaUI 和 ReactiveUI 框架中，Observable（可观察序列）是响应式编程的核心。它代表一个随时间推移的数据流，可以被订阅以接收数据更新。

1.1 Observable 的基本用法

using System;
using System.Reactive.Linq;

// 创建一个简单的 Observable
var observable = Observable.Return("Hello, Avalonia!");

// 订阅 Observable
var subscription = observable.Subscribe(
value => Console.WriteLine($"收到值: {value}"),
error => Console.WriteLine($"发生错误: {error}"),
() => Console.WriteLine("序列完成")
);

// 创建间隔 Observable（每秒发射一个值）
var timerObservable = Observable.Interval(TimeSpan.FromSeconds(1))
.Take(5); // 只取前5个值

timerObservable.Subscribe(
tick => Console.WriteLine($"Tick: {tick}")
);

1.2 在 AvaloniaUI 中创建 Observable 属性

using ReactiveUI;
using System.Reactive.Linq;

public class UserViewModel : ReactiveObject
{
private string _name;

[Reactive]
public string Name
{
get => _name;
set => this.RaiseAndSetIfChanged(ref _name, value);
}

// 创建一个基于 Name 属性的 Observable
public IObservable<string> NameObservable =>
this.WhenAnyValue(x => x.Name);

// 创建一个过滤后的 Observable（只包含长度大于3的名称）
public IObservable<string> ValidNameObservable =>
this.WhenAnyValue(x => x.Name)
.Where(name => !string.IsNullOrEmpty(name) && name.Length > 3);

public UserViewModel()
{
// 订阅 Name 变化
NameObservable.Subscribe(newName =>
{
Console.WriteLine($"用户名已更新: {newName}");
});

// 订阅有效名称变化
ValidNameObservable.Subscribe(validName =>
{
Console.WriteLine($"有效用户名: {validName}");
});
}
}

2. ObserveOn 的作用与用法

ObserveOn 是一个操作符，用于指定 Observable 序列在哪个调度器（Scheduler）上执行观察者（订阅者）的回调。在 UI 开发中，这通常用于确保 UI 更新在正确的线程上执行。

2.1 ObserveOn 的基本语法

using System.Reactive.Concurrency;
using System.Reactive.Linq;

// 创建 Observable
var source = Observable.Interval(TimeSpan.FromSeconds(1));

// 使用 ObserveOn 指定调度器
var uiObservable = source
.ObserveOn(RxApp.MainThreadScheduler) // 在 UI 线程观察
.Select(x => $"UI线程更新: {x}");

var backgroundObservable = source
.ObserveOn(Scheduler.Default) // 在后台线程观察
.Select(x => $"后台线程处理: {x}");

2.2 在 AvaloniaUI 中使用 ObserveOn

using Avalonia.Threading;
using ReactiveUI;
using System.Reactive.Concurrency;
using System.Reactive.Linq;

public class DataViewModel : ReactiveObject
{
private string _status;
private int _progress;

[Reactive]
public string Status
{
get => _status;
set => this.RaiseAndSetIfChanged(ref _status, value);
}

[Reactive]
public int Progress
{
get => _progress;
set => this.RaiseAndSetIfChanged(ref _progress, value);
}

public DataViewModel()
{
// 模拟后台数据流
var dataStream = Observable.Interval(TimeSpan.FromMilliseconds(100))
.Take(100) // 取100个值
.Select(x => (int)x * 10); // 转换为进度值

// 错误示例：直接在后台线程更新 UI 属性（可能导致跨线程异常）
// dataStream.Subscribe(progress => Progress = progress);

// 正确示例：使用 ObserveOn 确保在 UI 线程更新
dataStream
.ObserveOn(RxApp.MainThreadScheduler) // 关键：指定 UI 线程
.Subscribe(
progress =>
{
Progress = progress;
Status = $"处理中: {progress}%";
},
error => Status = $"错误: {error.Message}",
() => Status = "处理完成"
);
}

// 另一种方式：使用 Dispatcher
public async Task LoadDataAsync()
{
var dataStream = Observable.Interval(TimeSpan.FromMilliseconds(50))
.Take(50)
.Select(x => (int)x * 2);

await dataStream
.ObserveOn(RxApp.MainThreadScheduler)
.ForEachAsync(progress =>
{
Progress = progress;
});
}
}

3. Observable 与 ObserveOn 的核心区别

3.1 SubscribeOn vs ObserveOn

理解两者的区别对于正确使用 Reactive Extensions 至关重要：

using System.Reactive.Concurrency;
using System.Reactive.Linq;

// SubscribeOn：控制 Observable 的执行上下文（数据产生的线程）
var source1 = Observable.Create<int>(observer =>
{
Console.WriteLine($"Create 在线程: {Environment.CurrentManagedThreadId}");
observer.OnNext(1);
observer.OnCompleted();
return System.Reactive.Disposables.Disposable.Empty;
})
.SubscribeOn(NewThreadScheduler.Default); // 在新线程执行 Create

// ObserveOn：控制观察者的执行上下文（数据消费的线程）
var source2 = Observable.Return(2)
.ObserveOn(RxApp.MainThreadScheduler); // 在 UI 线程消费

// 组合使用
var combined = Observable.Interval(TimeSpan.FromSeconds(1))
.SubscribeOn(ThreadPoolScheduler.Instance) // 在线程池产生数据
.ObserveOn(RxApp.MainThreadScheduler) // 在 UI 线程消费数据
.Select(x => $"值: {x}");

4. 实际应用示例

4.1 实时搜索框实现

using ReactiveUI;
using System.Reactive.Linq;

public class SearchViewModel : ReactiveObject
{
private string _searchText;
private List<string> _results;

[Reactive]
public string SearchText
{
get => _searchText;
set => this.RaiseAndSetIfChanged(ref _searchText, value);
}

[Reactive]
public List<string> Results
{
get => _results;
set => this.RaiseAndSetIfChanged(ref _results, value);
}

public SearchViewModel()
{
// 创建搜索文本的 Observable
var searchObservable = this.WhenAnyValue(x => x.SearchText)
.Throttle(TimeSpan.FromMilliseconds(300)) // 防抖：300ms内只取最后一次
.DistinctUntilChanged() // 去重：值变化时才触发
.Where(text => !string.IsNullOrWhiteSpace(text))
.Select(text => text.Trim());

// 订阅搜索 Observable（在后台线程执行搜索，在 UI 线程更新结果）
searchObservable
.SelectMany(async text =>
{
// 模拟异步搜索（在后台线程执行）
await Task.Delay(100);
return SearchDatabase(text);
})
.ObserveOn(RxApp.MainThreadScheduler) // 切回 UI 线程更新结果
.Subscribe(results =>
{
Results = results;
});
}

private List<string> SearchDatabase(string keyword)
{
// 模拟数据库搜索
var data = new List<string>
{
"Apple", "Banana", "Cherry", "Date", "Elderberry",
"Fig", "Grape", "Honeydew", "Kiwi", "Lemon"
};

return data
.Where(item => item.Contains(keyword, StringComparison.OrdinalIgnoreCase))
.ToList();
}
}

4.2 多数据源合并与线程管理

using ReactiveUI;
using System.Reactive.Linq;

public class DashboardViewModel : ReactiveObject
{
private string _cpuUsage;
private string _memoryUsage;
private string _networkStatus;

[Reactive]
public string CpuUsage
{
get => _cpuUsage;
set => this.RaiseAndSetIfChanged(ref _cpuUsage, value);
}

[Reactive]
public string MemoryUsage
{
get => _memoryUsage;
set => this.RaiseAndSetIfChanged(ref _memoryUsage, value);
}

[Reactive]
public string NetworkStatus
{
get => _networkStatus;
set => this.RaiseAndSetIfChanged(ref _networkStatus, value);
}

public DashboardViewModel()
{
// 创建多个数据源 Observable
var cpuObservable = Observable.Interval(TimeSpan.FromSeconds(2))
.Select(_ => GetCpuUsage());

var memoryObservable = Observable.Interval(TimeSpan.FromSeconds(3))
.Select(_ => GetMemoryUsage());

var networkObservable = Observable.Interval(TimeSpan.FromSeconds(5))
.Select(_ => GetNetworkStatus());

// 合并所有 Observable，统一在 UI 线程更新
Observable.Merge(
cpuObservable.Select(usage => new { Type = "CPU", Value = usage }),
memoryObservable.Select(usage => new { Type = "Memory", Value = usage }),
networkObservable.Select(status => new { Type = "Network", Value = status })
)
.ObserveOn(RxApp.MainThreadScheduler) // 统一切换到 UI 线程
.Subscribe(data =>
{
switch (data.Type)
{
case "CPU":
CpuUsage = data.Value;
break;
case "Memory":
MemoryUsage = data.Value;
break;
case "Network":
NetworkStatus = data.Value;
break;
}
});
}

private string GetCpuUsage() => $"{new Random().Next(10, 90)}%";
private string GetMemoryUsage() => $"{new Random().Next(30, 95)}%";
private string GetNetworkStatus() => new Random().Next(0, 2) == 0 ? "在线" : "离线";
}

5. 常见问题与最佳实践

5.1 常见错误

• 忘记使用 ObserveOn：在后台线程更新 UI 属性导致跨线程异常
• 过度使用 ObserveOn：不必要的线程切换影响性能
• ObserveOn 位置错误：在错误的位置调用 ObserveOn 可能导致逻辑错误

5.2 最佳实践

5.3 性能优化示例

// 优化前：频繁更新导致性能问题
this.WhenAnyValue(x => x.SearchText)
.Subscribe(text => UpdateResults(text)); // 每次按键都触发

// 优化后：使用防抖和过滤
this.WhenAnyValue(x => x.SearchText)
.Throttle(TimeSpan.FromMilliseconds(300)) // 防抖
.DistinctUntilChanged() // 去重
.Where(text => text?.Length >= 3) // 过滤短文本
.ObserveOn(RxApp.MainThreadScheduler) // UI 线程更新
.Subscribe(text => UpdateResults(text));

6. 总结

Observable 和 ObserveOn 是 AvaloniaUI 响应式编程的两个核心概念：

• Observable 是数据流的抽象，用于表示随时间变化的数据序列
• ObserveOn 是调度器操作符，用于控制数据在哪个线程被消费
• 在 UI 开发中，ObserveOn(RxApp.MainThreadScheduler) 是确保线程安全的必要手段
• 正确理解和使用这两个概念，可以编写出更简洁、更安全、更高效的响应式 UI 代码

通过合理组合 Observable 的各种操作符（如 Select、Where、Throttle、Merge 等）和正确的线程调度，可以在 AvaloniaUI 中构建出响应迅速、代码清晰的数据驱动界面。

上周忘记了，这周加一篇