我刚接触ReactiveX,正在通过阅读源代码进行学习。一切都很清晰,但突然出现了一个名为“Consumer”的接口,它被用来代替Observer。
有人能告诉我它到底是做什么的吗?
我查了几个链接,但它们只说了一句话:Consumer是一个接受单个值的函数接口(回调)。
我想知道它的确切工作方式。
- 它是什么?
- 为什么需要它?
- 如何使用它?
- 它是否取代了Observer?如果是,怎么做以及为什么?
我刚接触ReactiveX,正在通过阅读源代码进行学习。一切都很清晰,但突然出现了一个名为“Consumer”的接口,它被用来代替Observer。
有人能告诉我它到底是做什么的吗?
我查了几个链接,但它们只说了一句话:Consumer是一个接受单个值的函数接口(回调)。
我想知道它的确切工作方式。
Consumer是一个简单的Java接口,它接受类型为T的变量。正如您所说,它用于回调。
示例:
import io.reactivex.functions.Consumer;
Flowable.just("Hello world").subscribe(new Consumer<String>() {
@Override public void accept(String s) {
System.out.println(s);
}
});
为什么它有效?我们如何使用Consumer代替Observer?
RxJava简单地创建了一个Observer,将Consumer传递给它,在onNext中调用它。
更新
LambdaObserver
是一种观察者,由四个功能接口创建,并将它们用作回调。它主要用于使用java 8 lambda表达式。它看起来像这样:
Observable.just(new Object())
.subscribe(
o -> processOnNext(o),
throwable -> processError(throwable),
() -> processCompletion(),
disposable -> processSubscription()
);
Consumer
是消费者,用于在订阅时消费接收到的值。它就像一个Subscriber
,将发出的数据作为回调函数传递。
Consumer
是一个简单的接口,具有通用类型的回调,并且需要接收Observable
发出的项目。
请注意,如果仅仅只有Consumer
,则不能捕获错误,调试时可能会出现问题。
您可以使用另一个Consumer作为第二个参数来解决这个问题,该Consumer将接收Throwable。
Flowable.just("Hello world")
.subscribe(
emittedData -> System.out.println(emittedData), // onNext
throwable -> throwable.printStackTrace() // onError
);
在我看来,Consumer 适用于反向/双向流。
例如,您的数据源通过 "热" 可观察对象 (HO) 发出一个复杂的时间依赖操作的 "Y",该操作是从参数 "X" 的形式执行的 "热" 可流式 (HF)。
假设参数 X 是通过 "热" 可观察对象 (HO) 发出的,那么您的数据源可以是一个 Consumer,它订阅 "HO" 并通过 HF 发出复杂操作的结果。
在这种情况下,您拥有双向流,并使用 Consumer 将数据源中通过 HO 提供的日期推送。
我不确定我的答案是否正确... Rx 有点复杂 :B