使用Flowable.generate RxJava处理错误

Question

我正在使用Flowable.generate构建一个Kafka消费者Flowable，下面是我的实现

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Flowable;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Properties;
public class RxKafkaUtils {
/**
* Returns a flowable impl for a Kafka Consumer
*/
public static @NonNull Flowable < Object > source(Properties props, String[] topics) {
final
var consumer = new KafkaConsumer(props);
return Flowable.generate(
() -> {
consumer.subscribe(Arrays.asList(topics));
return ConsumerRecords.empty().iterator();
},
(state, emitter) -> {
if (state.hasNext()) {
emitter.onNext(state.next());
return state;
} else {
try {
return consumer.poll(1000).iterator();
} catch (Throwable t) {
consumer.close();
emitter.onError(t);
throw t;           // <------ HERE
}
}
});
}
}

由于KafkaConsumer可以抛出Exceptions，所以我将.poll封装在try-catch中。

我的疑问是，我应该重新思考Throwable吗？如果没有，编译器会请求catch块的返回值，在这种情况下，我实际上没有任何

实现这一点的正确方法是什么(换句话说，优雅的关闭(？

Flowable.generate是正确的吗？我试着使用Flowable.fromPublisher，但状态管理完全一团糟，或者我做错了

Answer 1

好吧，原来

Flowable.generate接受在onComplete/onError之后被调用的disposeState消费者。当前状态(或我返回的最终状态传递给disposeState消费者(