Vert.x反应式Kafka客户端:写的时候链接不起作用？

Question

我使用的是io.vertx.reactivex.kafka.client.producer。KafkaProducer客户端。客户端有一个＜code＞rxWrite＜code＞函数，它返回＜code＞Single

我已经编写了以下工作示例。

test():测试链接和日志记录的函数

    fun test(): Single<Int> {
    val data = Single.just(ArrayList<String>().apply {
        add("Hello")
        add("World")
    })

    data.flattenAsObservable<String> { list -> list }
        .flatMap { advertiser ->
       //does not work with writeKafka
            writeError(advertiser).toObservable().doOnError({ println("Error $data") })
        }
        .subscribe({ record -> println(record) }, { e -> println("Error2 $e") })

    return data.map { it.size }
}

writeKafka：将给定的字符串写入Kafka并返回Single

fun writeKafka(param: String): Single<RecordMetadata> {
 //null topic to produce IllegalArgumentException()
    val record = KafkaProducerRecord.create(null, UUID.randomUUID().toString(), param)
    return kafkaProducer.rxWrite(record)
}

writeError：始终返回具有相同类型错误的单个

fun writeError(param: String): Single<RecordMetadata> {
    return Single.error<RecordMetadata>(IllegalArgumentException())
}

所以当我调用< code>writeKafka时，它只打印< code>Error2，但是如果我使用< code>writeError，它同时打印< code>Error和< code>Error2。看来< code>writeKafka返回的single还在等待结果，但是后来为什么连< code>Error2都打印出来了？

我是RxJava2中的新手，有人可以指出其中的任何错误吗？

Answer 1

阅读并发布错误堆栈非常重要，这样可以隔离问题。

在本例中，看起来您从＜code＞create＜code＞获得＜code＞IllegalArgumentException＜code＞并且您没有得到任何＜code＞单个返回kafkaProducer。rxWrite（record）根本不执行，实际上会使＜code＞平面图错误从不起作用，因此只打印“Error2”。