384
《C#並發編程經典實例》—— 發送通知給上下文
聲明：本文是《C#並發編程經典實例》的樣章，感謝圖靈授權並發編程網站發布樣章，禁止以任何形式轉載此文。
問題
Rx 盡量做到了線程不可知（thread agnostic）。因此它會在任意一個活動線程中發出通知（例如 OnNext）。
但是我們通常希望通知隻發給特定的上下文。例如 UI 元素隻能被它所屬的 UI 線程控製， 因此，如果要根據 Rx 的通知來修改 UI，就應該把通知“轉移”到 UI 線程。
解決方案
Rx 提供了 ObserveOn 操作符，用來把通知轉移到其他線程調度器。 看下麵的例子，使用 Interval，每秒鍾產生一個 OnNext 通知：
1 |
private void Button_Click(object sender, RoutedEventArgs e)
|
2 |
3 |
{ |
4 |
5 |
Trace.WriteLine("UI thread is " + Environment.CurrentManagedThreadId); Observable.Interval(TimeSpan.FromSeconds(1))
|
6 |
7 |
.Subscribe(x => Trace.WriteLine("Interval " + x + " on thread " + Environment.CurrentManagedThreadId));
|
8 |
9 |
} |
用我的電腦測試，顯示結果為：
UI thread is 9
Interval 0 on thread 10
Interval 1 on thread 10
Interval 2 on thread 11
Interval 3 on thread 11
Interval 4 on thread 10
Interval 5 on thread 11
Interval 6 on thread 11
因為 Interval 基於一個定時器（沒有指定的線程），通知會在線程池線程中引發，而不是 在 UI 線程中。要更新 UI 元素，可以通過 ObserveOn 輸送通知，並傳遞一個代表 UI 線程 的同步上下文：
01 |
private void Button_Click(object sender, RoutedEventArgs e)
|
02 |
03 |
{ |
04 |
05 |
var uiContext = SynchronizationContext.Current; |
06 |
07 |
Trace.WriteLine("UI thread is " + Environment.CurrentManagedThreadId); Observable.Interval(TimeSpan.FromSeconds(1))
|
08 |
09 |
.ObserveOn(uiContext) |
10 |
11 |
.Subscribe(x => Trace.WriteLine("Interval " + x + " on thread " + Environment.CurrentManagedThreadId));
|
12 |
13 |
} |
ObserveOn 的另一個常用功能是可以在必要時離開 UI 線程。假設有這樣的情況：鼠標一移
動，就意味著需要進行一些 CPU 密集型的計算。默認情況下，所有的鼠標移動事件都發 生在 UI 線程，因此可以使用 ObserveOn 把通知移動到一個線程池線程，在那裏進行計算， 然後再把表示結果的通知返回給 UI 線程：
private void Button_Click(object sender, RoutedEventArgs e)
{
var uiContext = SynchronizationContext.Current;
Trace.WriteLine(“UI thread is ” + Environment.CurrentManagedThreadId); Observable.FromEventPattern<MouseEventHandler, MouseEventArgs>(
handler => (s, a) => handler(s, a), handler => MouseMove += handler, handler => MouseMove -= handler)
.Select(evt => evt.EventArgs.GetPosition(this))
.ObserveOn(Scheduler.Default)
.Select(position =>
{
// 複雜的計算過程。
Thread.Sleep(100);
var result = position.X + position.Y; Trace.WriteLine(“Calculated result ” + result + ” on thread ” +
Environment.CurrentManagedThreadId);
return result;
})
.ObserveOn(uiContext)
.Subscribe(x => Trace.WriteLine(“Result ” + x + ” on thread ” + Environment.CurrentManagedThreadId));
}
運行這段代碼的話，就會發現計算過程是在線程池線程中進行的，計算結果在 UI 線程中
顯示。另外，還會發現計算和結果會滯後於輸入，形成等待的隊列，這種現象出現的原因 在於，比起 100 秒 1 次的計算，鼠標移動的更新頻率更高。Rx 中有幾種技術可以處理這 種情況，其中一個常用方法是對輸入流速進行限製，具體會在 5.4 節介紹。
討論
實際上，ObserveOn 是把通知轉移到一個 Rx 調度器上了。本節介紹了默認調度器（即線程 池）和一種創建 UI 調度器的方法。ObserveOn 最常用的功能是移到或移出 UI 線程，但調 度器也能用於別的場合。6.6 節介紹高級測試時，將再次關注調度器。
最後更新：2017-05-23 11:02:39