《C#并发编程经典实例》—— 用窗口和缓冲对事件分组

简介:

声明:本文是《C#并发编程经典实例》的样章,感谢图灵授权并发编程网站发布样章,禁止以任何形式转载此文。

问题

有一系列事件,需要在它们到达时进行分组。举个例子,需要对一些成对的输入作出响

应。第二个例子,需要在 2 秒钟的窗口期内,对所有输入进行响应。

解决方案

Rx 提 供 了 两 个 对 到 达 的 序 列 进 行 分 组 的 操 作:Buffer 和 Window。Buffer 会 留 住 到 达 的 事 件, 直 到 收 完 一 组 事 件, 然 后 会 把 这 一 组 事 件 以 一 个 集 合 的 形 式 一 次 性 地 转 送 过 去。 Window 会在逻辑上对到达的事件进行分组,但会在每个事件到达时立即传递过去。Buffer 的返回类型是 IObservable<IList<T>(由若干个集合组成的事件流);Window 的返回类型 是 IObservable<IObservable<T>(由若干个事件流组成的事件流)。

下面的例子使用 Interval,每秒创建 1 个 OnNext 通知,然后每 2 个通知做一次缓冲:


private void Button_Click(object sender, RoutedEventArgs e)

{

Observable.Interval(TimeSpan.FromSeconds(1))

.Buffer(2)

.Subscribe(x => Trace.WriteLine(

DateTime.Now.Second + ": Got " + x[0] + " and " + x[1]));

}


用我的电脑测试,每 2 秒产生 2 个输出:


13: Got 0 and 1

15: Got 2 and 3

17: Got 4 and 5

19: Got 6 and 7

21: Got 8 and 9

下面的例子有些类似,使用 Window 创建一些事件组,每组包含 2 个事件:

1

private void Button_Click(object sender, RoutedEventArgs e)

{

Observable.Interval(TimeSpan.FromSeconds(1))

.Window(2)

.Subscribe(group =>

{

});

}

Trace.WriteLine(DateTime.Now.Second + ": Starting new group");

group.Subscribe(

x => Trace.WriteLine(DateTime.Now.Second + ": Saw " + x),

() => Trace.WriteLine(DateTime.Now.Second + ": Ending group"));


用我的电脑测试,输出的结果就是这样:


17: Starting new group

18: Saw 0

19: Saw 1

19: Ending group

19: Starting new group

20: Saw 2

21: Saw 3

21: Ending group

21: Starting new group

22: Saw 4

23: Saw 5

23: Ending group

23: Starting new group


这几个例子说明了 Buffer 和 Window 的区别。Buffer 等待组内的所有事件,然后把所有事 件作为一个集合发布。Window 用同样的方法进行分组,但它是在每个事件到达时就发布。

Buffer 和 Window 都可以使用时间段作为参数。在下面的例子中,所有的鼠标移动事件被 收集进窗口,每秒一个窗口:


private void Button_Click(object sender, RoutedEventArgs e)

{

Observable.FromEventPattern<;MouseEventHandler, MouseEventArgs>;(

handler => (s, a) => handler(s, a), handler => MouseMove += handler, handler => MouseMove -= handler)

.Buffer(TimeSpan.FromSeconds(1))

.Subscribe(x => Trace.WriteLine(

DateTime.Now.Second + ": Saw " + x.Count + " items."));

}

输出的结果依赖于怎么移动鼠标,类似于这样:


49: Saw 93 items.

50: Saw 98 items.

51: Saw 39 items.

52: Saw 0 items.

53: Saw 4 items.

54: Saw 0 items.

55: Saw 58 items.


讨论

Buffer 和 Window 可用来抑制输入信息,并把输入塑造成我们想要的样子。另一个实用技 术是限流(throttling),将在 5.4 节介绍。

 

Buffer 和 Windows 都有其他重载,可用在更高级的场合。参数为 skip 和 timeShift 的重载 能创建互相重合的组,还可跳过组之间的元素。还有一些重载可使用委托,可对组的边界 进行动态定义。 

目录
相关文章
|
25天前
|
SQL 存储 算法
基于对象 - 事件模式的数据计算问题
基于对象-事件模式的数据计算是商业中最常见的数据分析任务之一。对象如用户、账号、商品等,通过唯一ID记录其相关事件,如操作日志、交易记录等。这种模式下的统计任务包括无序计算(如交易次数、通话时长)和有序计算(如漏斗分析、连续交易检测)。尽管SQL在处理无序计算时表现尚可,但在有序计算中却显得力不从心,主要原因是其对跨行记录运算的支持较弱,且大表JOIN和大结果集GROUP BY的性能较差。相比之下,SPL语言通过强化离散性和有序集合的支持,能够高效地处理这类计算任务,避免了大表JOIN和复杂的GROUP BY操作,从而显著提升了计算效率。
|
7月前
基于若依的ruoyi-nbcio流程管理系统支持支持定时边界事件和定时捕获事件
基于若依的ruoyi-nbcio流程管理系统支持支持定时边界事件和定时捕获事件
88 2
|
7月前
|
JavaScript 前端开发 Java
流的概念,怎么处理
流的概念,怎么处理
|
7月前
LabVIEW使用多个事件结构来处理同一个事件
LabVIEW使用多个事件结构来处理同一个事件
199 0
|
7月前
|
Java 数据处理 Apache
流计算中的窗口操作是什么?请解释其作用和使用场景。
流计算中的窗口操作是什么?请解释其作用和使用场景。
78 0
|
7月前
|
监控 Java 流计算
Flink中的窗口操作是什么?请解释其作用和使用场景。
Flink中的窗口操作是什么?请解释其作用和使用场景。
76 0
|
前端开发
前端学习笔记202305学习笔记第二十五天-读写流组合使用
前端学习笔记202305学习笔记第二十五天-读写流组合使用
34 0
第十二章队列模拟注意事项
第十二章队列模拟注意事项
59 0
|
存储 分布式计算 资源调度
有状态计算实现方式|学习笔记
快速学习有状态计算实现方式
有状态计算实现方式|学习笔记
|
缓存 Java vr&ar
【Java 网络编程】NIO Buffer 简介 ( 概念 | 数据传输 | 标记 | 位置 | 限制 | 容量 | 标记 | 重置 | 清除 | 翻转 | 重绕 | 链式操作 )
【Java 网络编程】NIO Buffer 简介 ( 概念 | 数据传输 | 标记 | 位置 | 限制 | 容量 | 标记 | 重置 | 清除 | 翻转 | 重绕 | 链式操作 )
202 0