RxJava - 主题的使用答案

【问题标题】：RxJava - Use of SubjectsRxJava - 主题的使用
【发布时间】：2017-10-29 13:14:50
【问题描述】：

我知道在 RxJava 世界中使用主题通常是不鼓励的。我试图观察可以通过主题轻松实现的对象的属性变化。我知道 rx.net 提供 Observable.FromEvent() 来轻松处理这种情况，但我想知道它们是否是基于非主题的 Java 替代方案。

【问题讨论】：

标签： rx-java rx-java2 reactivex

【解决方案1】：

您可以通过Observable.create 方法实现该功能：

Observable.create(emitter -> {
    PropertyObserver x = event -> emitter.onNext(event)
    yourObject.addObserver(x);
    // Set cancellable after adding observer, if the observable is
    //  disposed already it will properly remove the listener.
    emitter.setCancellable(() -> yourObject.removeObserver(x));
});

但您需要注意，每次使用此 observable 时，它都会为关联的对象添加一个新的侦听器。如果多个观察者依赖于对象的事件，则可以通过应用 share() 和 publish() 运算符来避免这种情况。

【讨论】：

【解决方案2】：

使用create 运算符。这里给你一个完整的例子。

package xdean.stackoverflow.rx;

import java.util.LinkedList;
import java.util.List;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.disposables.Disposable;

public class Q47000589 {
  public static void main(String[] args) {
    Property<Integer> p = new Property<>(1);
    Disposable d = p.listen().subscribe(i -> System.out.println("Listen: " + i));
    p.set(2);
    p.set(3);
    d.dispose();
    p.set(4);
    p.set(5);
  }

  public static class Property<T> {
    T value;
    List<ObservableEmitter<? super T>> emitters = new LinkedList<>();

    public Property(T value) {
      this.value = value;
    }

    T get() {
      return value;
    }

    void set(T t) {
      this.value = t;
      emitters.forEach(e -> e.onNext(t));
    }

    Observable<T> listen() {
      return Observable.create(e -> {
        emitters.add(e);
        e.onNext(value);
        e.setCancellable(() -> emitters.remove(e));
      });
    }
  }
}

输出是：

Listen: 1
Listen: 2
Listen: 3

【讨论】：