RxJava反应式编程

流控制和回压

｜

179

它必须在我们想要停止采样的时候完成。第二个流的完成标志着给定批次的结束。这里需

要仔细观察：我们想要开始一个新的批次时，

openings 流会发布一个事件。对于 openings

发布的每个事件，样例都会返回一个新的 Observable，它应该在未来的某个时间点完成。

例如，openings 发布一个值为 Duation.ofMillis(100) 的事件时，样例将其转换为一个

Observable，它会在给定的批次结束时完成，在这里也就是

100

毫秒之后。注意，在连续

的批次中，有些事件可能会被丢弃或重复出现。如果第二个 Observable，也就是负责标记

给定批次完成的 Observable，出现在下一个批次的开始事件之前，那么在这个时间差之内

的事件将会被 buffer() 丢弃。具体到这里：每秒（在业务时间之外是其他间隔秒数）都会

缓冲事件，但是缓冲区会在

100

毫秒（或者

200

毫秒，视情况而定）之后关闭，缓冲内容

会被转发至下游。大多数事件都会落在缓冲时段中，因此会被丢弃。

buffer()

操作符非常灵活和复杂。请你用它做一些试验性的尝试并确保理解了上述样例。

它能够非常智能地批量处理来自上游的事件，从而实现分组、采样和窗口移动的功能。但

是，

buffer()

关闭当前缓冲时需要创建一个中间的

List

，然后才能将其传递到下游，这可

能会给垃圾回收和内存使用带来不必要的压力（参见

8.6

节）。因此，

window()

操作符应运

而生。

6.1.3

　窗口移动

我们在使用 buffer() ...