【问题标题】:RxJava BehaviorSubject not emitting last item?RxJava BehaviorSubject 没有发出最后一项?
【发布时间】:2017-07-24 17:34:32
【问题描述】:

我有一个简单的 RxJava,使用 ReplaySubject,我可以得到打印所有 3 个数字的结果。

    val observable : Observable<Int> = Observable.just(1, 2, 3)
    val subject = ReplaySubject.create<Int>()
    observable.subscribe(subject)
    subject.subscribe{
        result ->
        System.out.println("Start $result in Subscription Result")
    }

当我更改为 Behavior 时,我希望打印第三个数字,即 3,因为我一直认为 Behavior 是重播最后一个发出的项目。

    val observable : Observable<Int> = Observable.just(1, 2, 3)
    val subject = BehaviorSubject.create<Int>()
    observable.subscribe(subject)
    subject.subscribe{
        result ->
        System.out.println("Start $result in Subscription Result")
    }

但是它不打印任何东西。为什么?

我在这里错过了什么重要的事情吗?如果是,请告诉我如何打印假设发出的最后一项(即 3)。

【问题讨论】:

  • BehaviorSubject 在按设计完成后不保留最后一项。您应该使用ReplaySubject.createWithSize(1) 始终重播替代测试,即使超出上游完成。
  • 谢谢@akarnokd。如果我只想发出最后一项,而不是全部,该怎么办?

标签: java kotlin rx-java rx-java2


【解决方案1】:

它不打印任何东西,因为订阅已经终止。如果订阅仍然有效,则会打印 3,例如:

val o1: Observable<Int> = Observable.just(1, 2, 3)
val o2: Observable<Int> = Observable.just(4).delay(100,TimeUnit.MILLISECONDS)
val observable: Observable<Int> = Observable.concat(o1, o2)
val subject = BehaviorSubject.create<Int>()
observable.subscribe(subject)
subject.subscribe{
    result ->
    System.out.println("Start $result in Subscription Result")
}
Thread.sleep(1000)

将打印 3 和 4(延迟后),其中 3 作为订阅前的最新事件发出,4 在订阅后发出。

此外,正如@akarnokd 在评论部分中所解释的那样,ReplaySubject.createWithSize(1) 可用于始终重播最后一个项目,即使在可观察完成之后,如果无论流完成状态如何都需要单个项目,则 observable.takeLast(1).subscribe(subject) 可用于保证:

val observable : Observable<Int> = Observable.just(1, 2, 3)
val subject = ReplaySubject.createWithSize<Int>(1)
observable.takeLast(1).subscribe(subject) //can be moved after subject.subscribe as well
subject.subscribe{
    result ->
    System.out.println("Start $result in Subscription Result")
}

【讨论】:

  • 是的,这就解释了为什么它不再有最后一项,因为它已经完成了。但是如果我想让最后一个项目仍然发射,有没有办法?
  • 正如@akarnokd 解释的那样,您可以使用ReplaySubject.createWithSize(1) 始终重播最后一项。
  • 它工作正常,只重播了 3 个。你可以粘贴你的代码吗?您确定 observable 在订阅时已终止吗? javadoc 明确声明了保证:“当观察者订阅终止的 ReplaySubject 时,他们保证最多看到 size onNext 事件,然后是终止事件。”
  • 根据我上面的第一个示例代码,我基本上将 val subject = ReplaySubject.create&lt;Int&gt;() 替换为 val subject = ReplaySubject.createWithSize&lt;Int&gt;(1)。
  • 不确定发生了什么,您使用的是哪个版本的 rx?无论如何,您可以保证单个事件,但使用 observable.takeLast(1).subscribe(subject) 修改 observable,无论订阅状态如何,这都将起作用。
猜你喜欢
  • 1970-01-01
  • 1970-01-01
  • 2019-08-31
  • 1970-01-01
  • 1970-01-01
  • 2018-11-07
  • 2018-11-18
  • 2021-11-01
  • 1970-01-01
相关资源
最近更新 更多