变换操作 - Buffer

优质
小牛编辑
129浏览
2023-12-01

Buffer

定期收集Observable的数据放进一个数据包裹,然后发射这些数据包裹,而不是一次发射一个值。

buffer

Buffer操作符将一个Observable变换为另一个,原来的Observable正常发射数据,变换产生的Observable发射这些数据的缓存集合。Buffer操作符在很多语言特定的实现中有很多种变体,它们在如何缓存这个问题上存在区别。

注意:如果原来的Observable发射了一个onError通知,Buffer会立即传递这个通知,而不是首先发射缓存的数据,即使在这之前缓存中包含了原始Observable发射的数据。

Window操作符与Buffer类似,但是它在发射之前把收集到的数据放进单独的Observable,而不是放进一个数据结构。

在RxJava中有许多Buffer的变体:

buffer(count)

buffer3

buffer(count)以列表(List)的形式发射非重叠的缓存,每一个缓存至多包含来自原始Observable的count项数据(最后发射的列表数据可能少于count项)

buffer(count, skip)

buffer4

buffer(count, skip)从原始Observable的第一项数据开始创建新的缓存,此后每当收到skip项数据,用count项数据填充缓存:开头的一项和后续的count-1项,它以列表(List)的形式发射缓存,取决于countskip的值,这些缓存可能会有重叠部分(比如skip < count时),也可能会有间隙(比如skip > count时)。

buffer(bufferClosingSelector)

buffer1

当它订阅原来的Observable时,buffer(bufferClosingSelector)开始将数据收集到一个List,然后它调用bufferClosingSelector生成第二个Observable,当第二个Observable发射一个TClosing时,buffer发射当前的List,然后重复这个过程:开始组装一个新的List,然后调用bufferClosingSelector创建一个新的Observable并监视它。它会一直这样做直到原来的Observable执行完成。

buffer(boundary)

buffer8

buffer(boundary)监视一个名叫boundary的Observable,每当这个Observable发射了一个值,它就创建一个新的List开始收集来自原始Observable的数据并发射原来的List

buffer(bufferOpenings, bufferClosingSelector)

buffer2

buffer(bufferOpenings, bufferClosingSelector)监视这个叫bufferOpenings的Observable(它发射BufferOpening对象),每当bufferOpenings发射了一个数据时,它就创建一个新的List开始收集原始Observable的数据,并将bufferOpenings传递给closingSelector函数。这个函数返回一个Observable。buffer监视这个Observable,当它检测到一个来自这个Observable的数据时,就关闭List并且发射它自己的数据(之前的那个List)。

buffer(timespan, unit[, scheduler])

buffer5

buffer(timespan, unit)定期以List的形式发射新的数据,每个时间段,收集来自原始Observable的数据(从前面一个数据包裹之后,或者如果是第一个数据包裹,从有观察者订阅原来的Observale之后开始)。还有另一个版本的buffer接受一个Scheduler参数,默认情况下会使用computation调度器。

buffer(timespan, unit, count[, scheduler])

buffer6

每当收到来自原始Observable的count项数据,或者每过了一段指定的时间后,buffer(timespan, unit, count)就以List的形式发射这期间的数据,即使数据项少于count项。还有另一个版本的buffer接受一个Scheduler参数,默认情况下会使用computation调度器。

buffer(timespan, timeshift, unit[, scheduler])

buffer7

buffer(timespan, timeshift, unit)在每一个timeshift时期内都创建一个新的List,然后用原始Observable发射的每一项数据填充这个列表(在把这个List当做自己的数据发射前,从创建时开始,直到过了timespan这么长的时间)。如果timespan长于timeshift,它发射的数据包将会重叠,因此可能包含重复的数据项。

还有另一个版本的buffer接受一个Scheduler参数,默认情况下会使用computation调度器。

buffer-backpressure

你可以使用Buffer操作符实现反压backpressure(意思是,处理这样一个Observable:它产生数据的速度可能比它的观察者消费数据的速度快)。

bp.buffer2

Buffer操作符可以将大量的数据序列缩减为较少的数据缓存序列,让它们更容易处理。例如,你可以按固定的时间间隔,定期关闭和发射来自一个爆发性Observable的数据缓存。这相当于一个缓冲区。

示例代码

  1. Observable<List<Integer>> burstyBuffered = bursty.buffer(500, TimeUnit.MILLISECONDS);

bp.buffer1

或者,如果你想更进一步,可以在爆发期将数据收集到缓存,然后在爆发期终止时发射这些数据,使用 Debounce 操作符给buffer操作符发射一个缓存关闭指示器(buffer closing indicator)可以做到这一点。

代码示例:

  1. // we have to multicast the original bursty Observable so we can use it
  2. // both as our source and as the source for our buffer closing selector:
  3. Observable<Integer> burstyMulticast = bursty.publish().refCount();
  4. // burstyDebounced will be our buffer closing selector:
  5. Observable<Integer> burstyDebounced = burstyMulticast.debounce(10, TimeUnit.MILLISECONDS);
  6. // and this, finally, is the Observable of buffers we're interested in:
  7. Observable<List<Integer>> burstyBuffered = burstyMulticast.buffer(burstyDebounced);

参见