java - 关于 RxJava 和 PublishSubject 的初学者问题

Question

我对 RxJava 中的 PublishSubject 有疑问。我创建了一个虚拟的 PublishSubject 来发出一些对象。这是我的代码:

override fun generate(exportRequest: ExportRequest): Observable<Report> {
        val faker = Faker()
        val dummyPublisher = PublishSubject.create<Report>()
        for(x in 1..1_000){
            val dataToExport = DataToExport(UUID.randomUUID(), faker.company().buzzword(), faker.company().name())
            val report = Report(dataToExport)
            sddPublisher.onNext(report)
            Thread.sleep(1)
        }
        dummyPublisher.onComplete()
        return dummyPublisher
    }

订阅时，不会发出任何对象。例如，没有打印任何内容:

... // somewhere in the code
reportStrategy.generate(exportRequest).subscribe { report: Report? ->
     println(report)
 }

也许我错过了一些东西。任何帮助将不胜感激

Answer 1

正如 @akarnokd 在评论中指出的，PublishSubject您创建的内容会立即发出 onNext 传递给它的任何值方法。无论当前是否订阅了任何内容，都会发生这种情况。它的设计主要是为了帮助弥合命令式或基于回调的代码与响应式(Reactive)代码之间的差距。

您似乎想要的是 Observable一旦有东西订阅它，它就开始执行一些同步代码。 Observable.create是创建此类实例的一种方法，但正确使用可能很麻烦。

创建您想要的内容的更方便的方法是 Observable.fromPublisher 。需要 Publisher作为一个论点。一个Publisher本身就是一个传递 Subscriber 的函数每当 Observer 时实例订阅Observable创建者 fromPublisher并允许您直接向 Observer 发送事件.

您想要的代码如下所示:

fun generateReportStream(genFakeReport: () -> Report): Observable<Report> {
    return Observable.fromPublisher { subscriber ->
        for (x in 1..1_000) {
            val fakeReport = genFakeReport()
            subscriber.onNext(fakeReport)
            Thread.sleep(1)
        }
        subscriber.onComplete()
    }
}

fun main() {
    /** supply whatever logic you want to generate a fake [Report] */
    fun genFakeReport(): Report = TODO()
    val subscription = generateReportStream(::genFakeReport).subscribe(::println)
}

一旦订阅 Observable，这将正确发出值。 generateReportStream 返回的实例。此外，可以对同一实例进行更多订阅，并且每个订阅都将使用相同的逻辑发出新的序列值。