1

当我将 Reactive Streams ( https://github.com/reactor/reactor-core ) 与自定义功能Publisher结合使用时publishOn，我总是得到一个 NPE。我的代码有什么问题？我是否Publisher以错误的方式使用？

Flux.from(MyPublisher())
            .publishOn(Schedulers.single())
            .subscribe { println("<-- $it received") }

class MyPublisher : Publisher<Int> {
    override fun subscribe(sub: Subscriber<in Int>) {
        while (true) {
            Thread.sleep(300)
            sub.onNext(1)
        }
    }
}

例外是：

Exception in thread "main" java.lang.NullPointerException
    at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.onNext(FluxPublishOn.java:212)
    at org.guenhter.kotlin.hello.MyPublisher.subscribe(HelloWorld.kt:18)
    at reactor.core.publisher.FluxSource.subscribe(FluxSource.java:52)
    at reactor.core.publisher.FluxPublishOn.subscribe(FluxPublishOn.java:96)
    at reactor.core.publisher.Flux.subscribe(Flux.java:6447)
    at reactor.core.publisher.Flux.subscribeWith(Flux.java:6614)
    at reactor.core.publisher.Flux.subscribe(Flux.java:6440)
    at reactor.core.publisher.Flux.subscribe(Flux.java:6404)
    at reactor.core.publisher.Flux.subscribe(Flux.java:6347)
    at org.guenhter.kotlin.hello.HelloWorldKt.main(HelloWorld.kt:11)
4

1 回答 1

3

Publisher由“reactive-streams”标准定义，并有许多要求。其中一个要求是Subscriber.onSubscribe必须在任何其他方法之前调用以遵循协议。

由于您还没有这样做，这意味着可能没有正确初始化某些东西，从而导致反应器类内部的 NPE。

但是，即使您解决了这个问题，该标准也被设计为反应式的，这意味着它仅在订阅者请求时才发出数据。因为您将向其发送数据，无论这可能会在以后导致异常。用于Flux.create创建可以正确处理请求的发射器，而不是创建您自己的 Publisher 实现。

于 2017-09-28T15:09:45.817 回答