java - 单个 Observable 是否按顺序通知其所有观察者？

Question

我编写了一个基于 PublishSubject 的 ObservableList。每次客户端向此列表添加一个元素时，观察者都会通过主题上的 onNext 方法收到通知。

我编写了两个示例观察者并将它们订阅到 ObservableList。我注意到通知是顺序的并且是阻塞的。因此，如果我订阅了这两个观察者，则会发生以下情况:

• 观察者收到通知的顺序与订阅的顺序相同。
• 如果一个 Observer 处于阻塞状态，那么下一个 Observer 只会在前一个 Observer 结束执行时收到通知。

我真的不在乎 id 调用是顺序的，但想了解为什么通知也被阻塞以及我怎样才能让它不被阻塞。

这是ObservableList

的代码

package co.com.subjects.example;

import java.util.ArrayList;
import java.util.List;

import rx.Observable;
import rx.functions.Action1;
import rx.subjects.PublishSubject;

public class ObservableList<T>{

    public String nombre;
    protected final List<T> list;
    protected final PublishSubject<T> onAdd;

    public ObservableList(String nombre) {
        this.list = new ArrayList<T>();
        this.onAdd = PublishSubject.create();
        this.nombre = nombre;
    }

    public void add(T value) {
        list.add(value);
        onAdd.onNext(value);
    }

    public Observable<T> getObservable() {
        return onAdd;
    }
}

Answer 1

want to understand why the notification is blocking

RxJava 的实现假设执行 Observer 的 onNext 方法总是快速且廉价的，所以 PublishSubject 只调用所有的 onNext 方法它的观察者一个接着一个，没有引入任何并发。

as well and how can I make it not to be blocking

在订阅您的观察者之前，您可以插入 .observeOn(Schedulers.computation())(或不同的调度程序，具体取决于您的需要)，以便 onNext调用在线程池上执行。