使用Reactivex的Consumer接口

Question

我刚接触ReactiveX，正在通过阅读源代码进行学习。一切都很清晰，但突然出现了一个名为“Consumer”的接口，它被用来代替Observer。

有人能告诉我它到底是做什么的吗？

我查了几个链接，但它们只说了一句话：Consumer是一个接受单个值的函数接口（回调）。

我想知道它的确切工作方式。

1. 它是什么？
2. 为什么需要它？
3. 如何使用它？
4. 它是否取代了Observer？如果是，怎么做以及为什么？

- Abhishek Kumar

---

一样的。我必须像这样订阅（消费者）。不明白吗？ - filthy_wizard

Answer 1

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中调用它。

更新

• 您调用Observable.subscribe(Consumer onNext)

• Observable.subscribe(Consumer onNext, Consumer onError, Action onComplete, Consumer onSubscribe)被调用

• LambdaObserver被创建

LambdaObserver是一种观察者，由四个功能接口创建，并将它们用作回调。它主要用于使用java 8 lambda表达式。它看起来像这样：

Observable.just(new Object())
                .subscribe(
                        o -> processOnNext(o),
                        throwable -> processError(throwable),
                        () -> processCompletion(),
                        disposable -> processSubscription()
                );

- Amaksoft

---

使用消费者接口时，如何进行处理？提供示例代码片段将有所帮助，谢谢。 - abby

Answer 2

Consumer是消费者，用于在订阅时消费接收到的值。它就像一个Subscriber，将发出的数据作为回调函数传递。

Consumer是一个简单的接口，具有通用类型的回调，并且需要接收Observable发出的项目。

请注意，如果仅仅只有Consumer，则不能捕获错误，调试时可能会出现问题。

您可以使用另一个Consumer作为第二个参数来解决这个问题，该Consumer将接收Throwable。

Flowable.just("Hello world")
  .subscribe(
            emittedData -> System.out.println(emittedData), // onNext
            throwable -> throwable.printStackTrace() // onError
);

- Emanuel

Answer 3

在我看来，Consumer 适用于反向/双向流。

例如，您的数据源通过 "热" 可观察对象 (HO) 发出一个复杂的时间依赖操作的 "Y"，该操作是从参数 "X" 的形式执行的 "热" 可流式 (HF)。

假设参数 X 是通过 "热" 可观察对象 (HO) 发出的，那么您的数据源可以是一个 Consumer，它订阅 "HO" 并通过 HF 发出复杂操作的结果。

在这种情况下，您拥有双向流，并使用 Consumer 将数据源中通过 HO 提供的日期推送。

我不确定我的答案是否正确... Rx 有点复杂 :B

- Felipe Porge Xavier