使用 Reactive Extensions - 基础知识
5.00/5 (12投票s)
如何连接热观察者到 Rx。
引言
当你开始编程时，你会遇到的第一件事就是事件，这些是人与你的程序交互所必需的东西。在本文中，我将介绍使用响应式扩展 (Rx) 与用户界面交互的最常见方法。本文将重点关注所谓的“热”可观察对象（hot observables），即使用 Rx 监听 UI 线程上已在运行的事件。我将简单地说明什么是热可观察对象和冷可观察对象（cold observables）；热可观察对象即使在没有人监听时也会发出值。冷可观察对象只有在你订阅它们后才活跃，就像连接到数据库一样。你可以观看 这个 视频以获得更深入的解释。
不幸的是，Rx 的文档和教程有些模糊，仅凭搜索很难挑选出最佳解决方案或最佳实践，因为在许多情况下，自一些博客撰写以来，rx-main 框架已经进行了多次更新。一件事情有很多种做法，因此找到适合特定实例的解决方案非常具有挑战性。这就是我决定写这篇文章的原因之一，这是我刚开始学习使用 Rx 时遇到的，它给我带来了一些我希望避免的挫折时刻。这就像我希望读到的教程一样。
如果你想从头开始使用 Rx 创建一个应用程序，你需要从 Microsoft 下载 Rx 库，它在 NuGet 上称为 Rx-main。我还建议你获取 Microsoft 发布的 Rx-xaml 包，因为它包含一些你以后需要使用的有用函数，尽管我在本文中没有使用其中的任何代码。
按钮点击事件
我将使用 Rx 来过滤鼠标和键盘输入，向你展示如何将它们连接到可观察流。我将首先在一个标准的 WPF 窗口中放置 2 个控件。
- Button
- 文本框
除了 Window 加载事件，我不会在 XAML 中直接挂钩任何事件，我将在代码隐藏中完成所有操作。我将从挂钩应用程序中最常见的事件——按钮点击事件开始本教程。
var buttonClick = Observable.FromEventPattern<RoutedEventHandler, RoutedEventArgs>(
h => btnMainWindow.Click += h,
h => btnMainWindow.Click -= h);
上面的代码实际上有几个地方值得注意。你可能首先注意到它有两个方法，一个用于向 Click 事件添加处理程序，另一个用于移除 Click 事件的处理程序。这是 Rx 的一个关键点，流实际上会知道如何自动取消订阅事件，这意味着你可以在不知道事件处理程序的情况下解除事件绑定。稍后我还会再谈到这一点。
你可能也猜到了，它通过一个名为 FromEventPattern 的函数创建了一个可观察事件流，并且它像标准的点击事件一样拥有 sender 和 EventArgs。我实际上在编写代码时有点懒，使用了 var 关键字，实际上返回到 buttonClick 的是这个
IObservable<EventPattern<RoutedEventArgs>> buttonClick =
Observable.FromEventPattern<RoutedEventHandler, RoutedEventArgs>(
h => btnMainWindow.Click += h,
h => btnMainWindow.Click -= h);
EventPattern 只是一个容器，允许你像这样从中获取 Sender 和 EventArgs。
var Sender = buttonClick.Select(evt => evt.Sender);
var EventArgs = buttonClick.Select(evt => evt.EventArgs);
实际上，FromEventPattern 非常通用，通过适当的修改，它可以让你监听属性更改或将属性描述符挂钩到属性。
我现在可以通过调用 buttonClick 中的 Subscribe 函数来监听事件。
IDisposable DisposableButton = buttonClick
.Subscribe(evt =>
{
var sender = evt.Sender;
var eventArgs = evt.EventArgs;
txtMainWindow.Text = "";
}
);
上面的代码中有一件非常重要的事情，那就是 subscribe 函数实际上返回一个 IDisposable。这实际上允许你通过调用 Dispose 方法来取消订阅，即移除处理程序。
DisposableButton.Dispose();
然而，在正常情况下，你可能有几个订阅，你想在窗口关闭或其他事件时终止它们。对于这些情况，存在一个称为 CompositeDisposable 的东西，我在 Window 中声明了它。
public partial class MainWindow : Window
{
CompositeDisposable DisposableSubscriptions = new CompositeDisposable();
public MainWindow()
{
InitializeComponent();
...
现在我可以将订阅添加到 CompositeDisposable 中了。
DisposableSubscriptions.Add(buttonClick
.Subscribe(evt =>
{
txtMainWindow.Text = "";
}
));
这使我能够通过一次调用关闭所有订阅，这非常方便。
private void Window_Closing(object sender, System.ComponentModel.CancelEventArgs e)
{
DisposableSubscriptions.Dispose();
}
乍一看，使用 Rx 可能并没有带来太多好处，你基本上只是得到了一个允许你在按钮点击时执行操作的子例程，它与普通的按钮点击 void 方法具有相同的能力，可以获取 sender 和 event arguments 来执行一些操作。Rx 的真正强大之处在于当你想要进行异步操作、弱订阅或基于用户输入的某些查询时。
还有一点是事件现在是一个流，它们被称为一流事件。这只是一个花哨的说法，告诉你你可以将整个流传递给不同的线程，并在那里进行订阅和取消订阅。
响应多个键盘输入
大多数应用程序都设计用来接收键盘输入以执行一些高级编辑。例如，TextBox 的常规功能，当你按下 Ctrl + C 进行复制，Ctrl + X 进行剪切等等。我的方法基本上是对 Stackoverflow 问题中的内容进行一些改进和泛化。
我们首先获取 PreviewKeyDown 和 PreviewKeyUp 事件，使用 Preview 事件的原因是基本上总能获取到应用程序中发生的键盘敲击，因为它们不会被事件冒泡或隧道阻碍。
var KeyDown = Observable.FromEventPattern<KeyEventArgs>(this, "PreviewKeyDown");
var KeyUp = Observable.FromEventPattern<KeyEventArgs>(this, "PreviewKeyUp");
我首先想要捕获的是按下左 Control 键，这很简单，除了一个小细节。默认情况下，当你按住一个键时，Windows 会发送多个键盘敲击，只要它被按下，但通过检查 IsRepeat 是否为 false 可以轻松避免这种情况。
var KeysAfterCtrlHold = from FirstKeyPress in KeyDown
.Where(e=> e.EventArgs.IsRepeat == false)
.Where(e=> e.EventArgs.Key == Key.LeftCtrl)
这段代码是标准的 Rx 操作，代码的真正之美在于用户在按下左 Control 键时按下更多的键盘按键。
var KeysAfterCtrlHold = from FirstKeyPress in KeyDown
.Where(e=> e.EventArgs.IsRepeat == false)
.Where(e=> e.EventArgs.Key == Key.LeftCtrl)
from RestOfKeyDown in KeyDown
.TakeUntil(KeyUp.Where(e => e.EventArgs.Key == Key.LeftCtrl))
.Where(e => e.EventArgs.Key != Key.LeftCtrl)
.Select(evt=>evt.EventArgs.Key)
select RestOfKeyDown;
在 RestOfKeyDown 中，我想要所有在左 Control 键按下的键盘敲击，但不想要左 Control 键的事件。另外，我只想要实际的按键事件，并且只将 Key 作为参数发送给订阅者，这是通过使用 .Select(evt=>evt.EventArgs.Key). 实现的。我在过滤之前无法做到这一点，因为我需要 EventArgs 来检查 IsRepeat 属性。最后一个 Select 语句基本上是告诉我，我只想要在左 Control 键按下之后发送的按键，而不是 Control 键按下事件。
到目前为止，看起来我付出了很多努力，但回报却不多。“关键”之处在于，代码是完全通用的。它会在左 Control 键按下时发送所有按键，因此我只需进行一些过滤就可以响应多个不同的按键序列。显而易见的做法是使用 Buffer 或 Window 来获取你感兴趣的序列。
如果我编写一个简单的订阅来响应按键组合 Ctrl + K + C （就像 Visual Studio 使用热键注释选定的代码一样），我将选择调用一个大小为 2 的 Buffer。（Buffer 是一个滑动窗口，你可以指定它应该存储的数量以及更新缓冲区所需的步数，调用 Buffer(3,1) 将存储 3 个连续值，并在 1 个新值进来时更新它们）。我现在可以编写一些简单的代码来获取按键。你应该注意到 Buffer 代码的返回值是一个 IList<T>，来自 Observable<T> 流，我将在稍后利用这一点。
IDisposable DisposableCtrlBuffer = KeysAfterCtrlHold.Buffer(2,1).Subscribe(evt =>
{
var key1 = evt[0];
var key2 = evt[1];
txtMainWindow.Text += Environment.NewLine + "Key " + key1.ToString() + " and key " + key2.ToString() + " is pressed.";
if (key1 == Key.K && key2 == Key.C)
{
txtMainWindow.Focus();
txtMainWindow.SelectAll();
this.Focus();
}
if (key1 == Key.K && key2 == Key.U)
{
txtMainWindow.Focus();
txtMainWindow.Select(0, 0);
this.Focus();
}
});
这很好，但实际上存在几个问题。首先，即使在 Ctrl 键释放后，它也会将 Buffer 存储在内存中。它也只会组合最后两次按下的按键，如果我在另一个订阅中选择监听 3 次按键输入，这可能会有问题。所以实际上我只想要按下 Ctrl 键后的前两次按键，并且一旦 Ctrl 键释放就重置 Buffer。
这实际上并不像你想象的那么难，我只需要以稍有不同的方式重写 Buffer 代码。与其指定缓冲区应包含多少个数字的按键和/或更新所需的跳过值数量，不如使用 文档中描述的一些重载。
在很多情况下，有趣的是形式为 Buffer(Start_A_New_Buffer, Close_The_Buffer) 的重载，可以这样使用：
.Buffer(
// Start a new buffer when left Ctrl is pressed
KeyDown.Where(e => e.EventArgs.Key == Key.LeftCtrl)
.Where(e=> e.EventArgs.IsRepeat == false),
// Close the buffer if left Ctrl is released
_ => KeyUp.Where(e => e.EventArgs.Key == Key.LeftCtrl));
给你一个小警告，它会包含开始时的 LeftCtrl 键信息。这不算坏事，但它不会发送值，直到 Close_The_Buffer 发生，在这种情况下是 LeftCtrl 键被释放。
另一个重载是仅指定何时关闭 Buffer，请注意，一旦出现关闭条件，它就会发送当前缓冲区。
Buffer( () => Buffer_close_condition)
这是我想在应用程序中使用的，但我有一个问题。我希望缓冲区在包含 2 个值后关闭，但也希望在释放 LeftCtrl 键时刷新它。我可以通过使用 Observable.Merge 函数来实现这一点：
var FlushableKeyBuffer = KeysAfterCtrlHold
.Buffer(() => Observable.Merge(KeysAfterCtrlHold.Buffer(2), CloseBufferOnCtrlUp.Buffer(1)))
.Where(evt=>evt.Count==2);
它基本上只是将两个单独的流合并成一个，所以在上面的代码中，它将传播 Ctrl 键按下后的按键，并在 Ctrl 键抬起事件发生时切换源。Buffer(1,1) 调用只是为了让两个函数在函数内部有相同的返回值。合并但我对传播 Ctrl 键抬起事件不感兴趣，所以我想阻止它被发送到订阅者。
这实际上非常有趣，正如我之前谈到的按键如何被视为热事件一样，我现在创建了一个过滤器，我可以将其描述为从热切换到冷，因为代码只在 Ctrl 键按下时才流式传输事件，而不是之前。
更新后的订阅代码现在看起来是这样的：
IDisposable DisposableCtrlBuffer = FlushableKeyBuffer.Subscribe(evt =>
{
var key1 = evt[0];
var key2 = evt[1];
txtMainWindow.Text += Environment.NewLine + "Key " + key1.ToString() + " and key " + key2.ToString() + " is pressed.";
if (key1 == Key.K && key2 == Key.C)
{
txtMainWindow.Focus();
txtMainWindow.SelectAll();
this.Focus();
}
if (key1 == Key.K && key2 == Key.U)
{
txtMainWindow.Focus();
txtMainWindow.Select(0, 0);
this.Focus();
}
});
这很不错，但有一种更简洁的方法来实现 KeysAfterCtrlHold，即使用 Observable.Join。它还展示了 Join 操作符是如何工作的，所以话不多说：
var KeysAfterCtrlHold2 = Observable.Join(
// Start the Left Control is down window
KeyDown.Where(evt => evt.EventArgs.Key == Key.LeftCtrl && evt.EventArgs.IsRepeat == false),
// Start listnening for key down events that isnt left control
KeyDown.Where(evt => evt.EventArgs.Key != Key.LeftCtrl && evt.EventArgs.IsRepeat == false),
// Close the Left Control window
s => KeyUp.Where(evt => evt.EventArgs.Key == Key.LeftCtrl && evt.EventArgs.IsRepeat == false),
// Make the key down event a point event
c => Observable.Empty<Key>(),
// The return key (s is always Left Control)
(s, c) => {return c.EventArgs.Key;}
);
功能完全相同，但可读性大大提高。我现在可以解释 Join 实际上做了什么，因为它有两个窗口来检查并发性。这只是一个花哨的说法，意味着 Left Ctrl 按钮与按下其他键同时按下。前两个方法调用告诉你两个窗口何时打开，第三个和第四个告诉你何时关闭。我选择尽快关闭另一个按键，因为我只需要它作为一个瞬时事件。正如你已经注意到的，如果其中一个窗口关闭，Join 就不会返回任何值。
代码还有另一个问题，它会发送 KeysAfterCtrlHold 中所有被延迟的值。原因是第一个 Join 会发送 Ctrl 键按下期间的所有按键。当它达到 2 个值或你释放 Ctrl 按钮时，缓冲区将开始发送值。缓冲区将在其中一个关闭参数发生时开始生成一个新的缓冲区，但这次，它不会在达到 2 个新值时重新释放，而是等待 Observable.Merge 中的第二个事件发生，即左 Ctrl 键抬起事件。这意味着如果你按住左 Ctrl 键并按了 10 次 A 键，它将首先按正常方式发送预期的 2 个 A 键。但如果你现在释放左 Ctrl 按钮，缓冲区将释放它存储的所有记录，所以你会得到 8 个 A 键。解决方案是停止发送 FlushableBuffer，只要 Ctrl 键没有被按下，我就创建了另一个 Observable.Join，它将在 Ctrl 键按下时传递缓冲区。
var BufferedKeysAfterCtrlHold = Observable.Join
(
// Start the Left Control is down window
KeyDown
.Where(evt => evt.EventArgs.Key == Key.LeftCtrl)
.Where(evt=> evt.EventArgs.IsRepeat == false),
// Start listnening for key down events that isnt left control
FlushableKeyBuffer,
// Close the Left Control window
s => KeyUp.Where(evt => evt.EventArgs.Key == Key.LeftCtrl),
// Make the key down event a point event
c => Observable.Empty<Key>(),
// The return key (s is always Left Control)
(s, c) => { return c; }
);
为了可重用性，我将所有步骤合并到一个函数调用中：
private IObservable<System.Collections.Generic.IList<Key>> Keystrokes(IObservable<EventPattern<KeyEventArgs>> KeyDown, IObservable<EventPattern<KeyEventArgs>> KeyUp, Key WhileThisKeyIsDown, int NumberOfKeysPushed)
{
var GetKeysAfter = Observable.Join(
// Start the Left Control is down window
KeyDown.Where(evt => evt.EventArgs.Key == WhileThisKeyIsDown && evt.EventArgs.IsRepeat == false),
// Start listnening for key down events that isnt left control
KeyDown.Where(evt => evt.EventArgs.Key != WhileThisKeyIsDown && evt.EventArgs.IsRepeat == false),
// Close the Left Control window
s => KeyUp.Where(evt => evt.EventArgs.Key == WhileThisKeyIsDown),
// Make the key down event a point event
c => Observable.Empty<Key>(),
// The return key (s is always Left Control)
(s, c) => { return c.EventArgs.Key; }
);
var CloseTheBufferOnKeyUp = KeyUp.Where(e => e.EventArgs.Key == WhileThisKeyIsDown)
.Select(x => x.EventArgs.Key);
var FlushableKeyBuffer = GetKeysAfter
.Buffer(() => Observable.Merge(GetKeysAfter.Buffer(NumberOfKeysPushed, 1), CloseTheBufferOnKeyUp.Buffer(1, 1)))
.Where(evt => evt.Count == NumberOfKeysPushed);
var FilteredBufferedKeys = Observable.Join(
// Start the Left Control is down window
KeyDown.Where(evt => evt.EventArgs.Key == WhileThisKeyIsDown && evt.EventArgs.IsRepeat == false),
// Start listnening for key down events that isnt left control
FlushableKeyBuffer,
// Close the Left Control window
s => KeyUp.Where(evt => evt.EventArgs.Key == WhileThisKeyIsDown),
// Make the key down event a point event
c => Observable.Empty<Key>(),
// The return key (s is always Left Control)
(s, c) => { return c; }
);
return FilteredBufferedKeys;
}
合并鼠标输入
这段代码很大程度上基于 Channel 9 视频关于 Observable.Join 中的一个挑战，我必须说，Join 操作符对我来说很难理解，直到我看到名为 Programming Streams of Coincidence with Join and GroupJoin for Rx 的 Channel 9 视频。有了解释以及其他程序员发布的示例，我终于明白了。
我基本想做的是结合 MouseDown 和 MouseUp 事件，并发布点击的持续时间，所以我首先获取事件流：
var MouseDown = Observable.FromEventPattern<MouseButtonEventArgs>(this, "MouseDown");
var MouseUp = Observable.FromEventPattern<MouseButtonEventArgs>(this, "MouseUp");
完整的代码相当简短，我只是获取 MouseDown 和 MouseUp 事件，并根据按下左键的输入进行过滤。Select 可能更有趣，因为 _ => 表示接收来自鼠标事件的任何参数，并返回当前的 DateTime。
var LeftMouseButtonPressedDuration = Observable.Join(
MouseDown
.Where(evt => evt.EventArgs.ChangedButton == MouseButton.Left)
.Select(_ => {return DateTime.Now;}),
MouseUp
.Where(evt => evt.EventArgs.ChangedButton == MouseButton.Left)
.Select(_ =>
{
DateTime now = DateTime.Now;
return now;
}
),
CloseMouseDownWindow => MouseUp,
MouseUpEvent => Observable.Empty<DateTime>(),
(TimeDown, TimeUp) =>
{
TimeSpan MousePressedDuration = TimeUp - TimeDown;
return "Total duartion is: " + MousePressedDuration.ToString(@"ss\:fff");
}
);
Join 操作符中有趣的部分发生在 CloseMouseDownWindow 事件，它告诉 Join 窗口何时结束。你实际上可以将其设置为 TimeDown => Observable.Never<DateTime>()，这将把所有发生的 MouseDown 事件存储到一个永远不会关闭的 Observable<T> 中。TimeUp 将告诉何时关闭窗口，通过将其设置为 => Observable.Empty，它将在事件发生时立即停止向上窗口。
这也是让我困惑的一点，s 和 c 在此时会返回单个值，但可以是 Observable<T> 流类型。所以当我得到结果 (s, c) => ... 时，它们的类型是 <T>。
最后剩下的是将结果发布到 WPF 窗口上的文本框中。
DisposableSubscriptions.Add(LeftMouseButtonPressedDuration.Subscribe(evt =>
{
txtMainWindow.Text += Environment.NewLine + evt.ToString();
})
);