0

我有需要解析字符串的应用程序，我使用它 RxScala。我的代码：

import java.util.concurrent.TimeUnit
import rx.lang.scala.Subject
import rx.lang.scala.schedulers.NewThreadScheduler
import rx.lang.scala.subjects.{SerializedSubject, PublishSubject}
import scala.concurrent.duration.Duration

object RxScala extends App {
  val subject: Subject[String] = SerializedSubject(PublishSubject())

  val processLines = (lines: Seq[String]) => {
    // long action
  }

  subject
    .subscribeOn(NewThreadScheduler())
    .tumblingBuffer(Duration(2, TimeUnit.SECONDS), 100)
    .subscribe(processLines)

  for(i <- 1 to 100000) {
    subject.onNext("Line " + i)
  }
}

我有问题，因为我添加行的速度更快，然后我可以处理它们。

我想创建缓冲区，例如 200 行，如果缓冲区已满，则在缓冲区未满时忽略新记录，例如

Add 100 records (A)
Add 100 records (B)
Program start processLines (A) // buffer have (B) elements
Add 100 records (C) // buffer have (B, C) elements and it is full
Add 100 records (D) // elements are ignored
ProcessLines is finished 
Program start processLines (B) // buffer have (C) elements
Add 100 records (E) // buffer have (C, E) elements

RxScala 有办法做到这一点吗？

4

1 回答 1

0 
subject.
    tumblingBuffer(Duration(2, TimeUnit.SECONDS), 100).
    onBackpressureDrop.
    observeOn(NewThreadScheduler()).
    subscribe(processLines)

http://reactivex.io/documentation/operators/backpressure.html

onBackpressureDrop 丢弃源 Observable 的排放，除非有来自下游订阅者的未决请求，在这种情况下，它将发出足够的项目来满足请求

于 2015-07-16T12:40:20.900 回答