Rxjava Subject分析

Posted 2020-07-13 一叶飘舟

Subject = Observable + Observer

看看官方的描述：

Subject可以看成是一个桥梁或者代理，在某些ReactiveX实现中（如RxJava），它同时充当了Observer和Observable的角色。因为它是一个Observer，它可以订阅一个或多个Observable；又因为它是一个Observable，它可以转发它收到(Observe)的数据，也可以发射新的数据。

由于一个Observable订阅一个Observable，它可以触发这个Observable开始发射数据（如果那个Observable是”冷”的–就是说，它等待有订阅才开始发射数据）。因此有这样的效果，Subject可以把原来那个”冷”的Observable变成”热”的。

通俗的理解就是：

subject是一个神奇的对象，它可以是一个Observable同时也可以是一个Observer：它作为连接这两个世界的一座桥梁。一个Subject可以订阅一个Observable，就像一个观察者，并且它可以发射新的数据，或者传递它接受到的数据，就像一个Observable。很明显，作为一个Observable，观察者们或者其它Subject都可以订阅它。

一旦Subject订阅了Observable，它将会触发Observable开始发射。如果原始的Observable是“冷”的，这将会对订阅一个“热”的Observable变量产生影响。

Subject的种类

针对不同的场景一共有四种类型的Subject。他们并不是在所有的实现中全部都存在，而且一些实现使用其它的命名约定（例如，在RxScala中Subject被称作PublishSubject）。

RxJava提供的四种不同的Subject：

1.PublishSubject

PublishSubject仅会向Observer释放在订阅之后Observable释放的数据。

官方介绍：

PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者。需要注意的是，PublishSubject可能会一创建完成就立刻开始发射数据（除非你可以阻止它发生），因此这里有一个风险：在Subject被创建后到有观察者订阅它之前这个时间段内，一个或多个数据可能会丢失。如果要确保来自原始Observable的所有数据都被分发，你需要这样做：或者使用Create创建那个Observable以便手动给它引入”冷”Observable的行为（当所有观察者都已经订阅时才开始发射数据），或者改用ReplaySubject。

如果原始的Observable因为发生了一个错误而终止，BehaviorSubject将不会发射任何数据，只是简单的向前传递这个错误通知。

2.BehaviorSubject

当Observer订阅了一个BehaviorSubject，它一开始就会释放Observable最近释放的一个数据对象，当还没有任何数据释放时，它则是一个默认值。接下来就会释放Observable释放的所有数据。如果Observable因异常终止，BehaviorSubject将不会向后续的Observer释放数据，但是会向Observer传递一个异常通知。

官方介绍：

当观察者订阅BehaviorSubject时，它开始发射原始Observable最近发射的数据（如果此时还没有收到任何数据，它会发射一个默认值），然后继续发射其它任何来自原始Observable的数据。

然而，如果原始的Observable因为发生了一个错误而终止，BehaviorSubject将不会发射任何数据，只是简单的向前传递这个错误通知。

3.ReplaySubject

不管Observer何时订阅ReplaySubject，ReplaySubject会向所有Observer释放Observable释放过的数据。
有不同类型的ReplaySubject，它们是用来限定Replay的范围，例如设定Buffer的具体大小，或者设定具体的时间范围。
如果使用ReplaySubject作为Observer，注意不要在多个线程中调用onNext、onComplete和onError方法，因为这会导致顺序错乱，这个是违反了Observer规则的。

官方介绍：

ReplaySubject会发射所有来自原始Observable的数据给观察者，无论它们是何时订阅的。也有其它版本的ReplaySubject，在重放缓存增长到一定大小的时候或过了一段时间后会丢弃旧的数据（原始Observable发射的）。

如果你把ReplaySubject当作一个观察者使用，注意不要从多个线程中调用它的onNext方法（包括其它的on系列方法），这可能导致同时（非顺序）调用，这会违反Observable协议，给Subject的结果增加了不确定性。

4.AsyncSubject

AsyncSubject仅释放Observable释放的最后一个数据，并且仅在Observable完成之后。然而如果当Observable因为异常而终止，AsyncSubject将不会释放任何数据，但是会向Observer传递一个异常通知。

官方介绍：

一个AsyncSubject只在原始Observable完成后，发射来自原始Observable的最后一个值。（如果原始Observable没有发射任何值，AsyncObject也不发射任何值）它会把这最后一个值发射给任何后续的观察者。

然而，如果原始的Observable因为发生了错误而终止，AsyncSubject将不会发射任何数据，只是简单的向前传递这个错误通知。

小结

回顾上面所讲的，不难看出不同的Subject最大的区别在于发送数据的行为不同，简单概括如下：

Subject	发射行为
AsyncSubject	不论订阅发生在什么时候，只会发射最后一个数据
BehaviorSubject	发送订阅之前一个数据和订阅之后的全部数据
ReplaySubject	不论订阅发生在什么时候，都发射全部数据
PublishSubject	发送订阅之后全部数据

关于Subject更详细的使用方法请查阅官方文档。

RxJava的对应类

假设你有一个Subject，你想把它传递给其它的代理或者暴露它的Subscriber接口，你可以调用它的asObservable方法，这个方法返回一个Observable。具体使用方法可以参考Javadoc文档。

如果你把 Subject 当作一个 Subscriber 使用，注意不要从多个线程中调用它的onNext方法（包括其它的on系列方法），这可能导致同时（非顺序）调用，这会违反Observable协议，给Subject的结果增加了不确定性。

要避免此类问题，你可以将 Subject 转换为一个 SerializedSubject ，类似于这样：

mySafeSubject = new SerializedSubject( myUnsafeSubject );

---

通过BehaviorSubject 来制作缓存

代码大致形式：

api.getData().subscribe(behaviorSubject); // 判断cache为空则获取数据，网络数据会被缓存
behaviorSubject.subscribe(observer);// 之前的缓存将直接送达observer

详细代码：

BehaviorSubject<List<Item>> cache;
public Subscription subscribeData(@NonNull Observer<List<Item>> observer) {
         //判断内存缓存是否为空
        if (cache == null) {
            cache = BehaviorSubject.create();
            Observable.create(new Observable.OnSubscribe<List<Item>>() {
                @Override
                public void call(Subscriber< ? super List<Item>> subscriber) {
                    List<Item> items = Database.getInstance().readItems();
                    //判断硬盘缓存是否为空
                    if (items == null) {
                        //从网络读取数据
                        loadFromNetwork();
                    } else {
                        //发送硬盘数据
                        subscriber.onNext(items);
                    }
                }
            }).subscribeOn(Schedulers.io())
              .subscribe(cache);
        } 
        return cache.observeOn(androidSchedulers.mainThread()).subscribe(observer);
    }

  subscription = subscribeData(new Observer<List<Item>>() {
                    @Override
                    public void onCompleted() {
                    }

                    @Override
                    public void onError(Throwable e) {
                        swipeRefreshLayout.setRefreshing(false);
                        Toast.makeText(getActivity(), R.string.loading_failed, Toast.LENGTH_SHORT).show();
                    }

                    @Override
                    public void onNext(List<Item> items) {
                        swipeRefreshLayout.setRefreshing(false);
                        adapter.setItems(items);
                    }
                });

---

通过PublishSubject实现传统的Observable Hello World

PublishSubject<String> stringPublishSubject = PublishSubject.create();
Subscription subscriptionPrint = stringPublishSubject.subscribe(new Observer<String>() {
    @Override
    public void onCompleted() {
        System.out.println("Observable completed");
    }

    @Override
    public void onError(Throwable e) {
        System.out.println("Oh,no!Something wrong happened!");                
    }

    @Override
    public void onNext(String message) {
        System.out.println(message);
    }
});
stringPublishSubject.onNext("Hello World");

在刚才的例子中，我们创建了一个PublishSubject，用create()方法发射一个String值，然后我们订阅了PublishSubject。此时，没有数据要发送，因此我们的观察者只能等待，没有阻塞线程，也没有消耗资源。就在这随时准备从subject接收值，如果subject没有发射值那么我们的观察者就会一直在等待。再次声明的是，无需担心：观察者知道在每个场景中该做什么，我们不用担心什么时候是因为它是响应式的：系统会响应。我们并不关心它什么时候响应。我们只关心它响应时该做什么。

最后一行代码展示了手动发射字符串“Hello World”,它触发了观察者的onNext()方法，让我们在控制台打印出“Hello World”信息。

让我们看一个更复杂的例子。话说我们有一个private声明的Observable，外部不能访问。Observable在它生命周期内发射值，我们不用关心这些值，我们只关心他们的结束。

首先，我们创建一个新的PublishSubject来响应它的onNext()方法，并且外部也可以访问它。

final PublishSubject<Boolean> subject = PublishSubject.create();

subject.subscribe(new Observer<Boolean>() {
    @Override
    public void onCompleted() {

    }

    @Override
    public void onError(Throwable e) {

    }

    @Override
    public void onNext(Boolean aBoolean) {
        System.out.println("Observable Completed");
    }
});

然后，我们创建“私有”的Observable，只有subject才可以访问的到。

Observable.create(new Observable.OnSubscribe<Integer>() {
    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        for (int i = 0; i < 5; i++) {
            subscriber.onNext(i);
        }
        subscriber.onCompleted();
    }
}).doOnCompleted(new Action0() {
    @Override
    public void call() {
        subject.onNext(true);
    }
}).subscribe();

Observable.create()方法包含了我们熟悉的for循环，发射数字。doOnCompleted()方法指定当Observable结束时要做什么事情：在subject上发射true。最后，我们订阅了Observable。很明显，空的subscribe()调用仅仅是为了开启Observable，而不用管已发出的任何值，也不用管完成事件或者错误事件。为了这个例子我们需要它像这样。

在这个例子中，我们创建了一个可以连接Observables并且同时可被观测的实体。当我们想为公共资源创建独立、抽象或更易观测的点时，这是极其有用的。

---

RxJava的Subject源码分析

package rx.subjects;

import rx.Observable;
import rx.Observer;
import rx.Subscriber;

/**
 * Represents an object that is both an Observable and an Observer.
 */
public abstract class Subject<T, R> extends Observable<R> implements Observer<T> {
    protected Subject(OnSubscribe<R> onSubscribe) {
        super(onSubscribe);
    }

    public abstract boolean hasObservers();

    public final SerializedSubject<T, R> toSerialized() {
        return new SerializedSubject<T, R>(this);
    }
}

---

BehaviorSubject源码分析

BehaviorSubject订阅subscribe过程

在需要使用subject时，调用Subject的subscribe(..)方法，该方法实际会调用下面这个subscribe(Subscriber< ? super T> subscriber)方法，所以其他的subscribe方法都要将输入参数转化为一个Subscriber对象。

public final Subscription subscribe(Subscriber<? super T> subscriber) {
        ...  
        // new Subscriber so onStart it
        subscriber.onStart();

        ...

        // The code below is exactly the same an unsafeSubscribe but not used because it would add a sigificent depth to alreay huge call stacks.
        try {
            // allow the hook to intercept and/or decorate
            hook.onSubscribeStart(this, onSubscribe).call(subscriber);
            return hook.onSubscribeReturn(subscriber);
        } catch (Throwable e) {
            // special handling for certain Throwable/Error/Exception types
            Exceptions.throwIfFatal(e);
            // if an unhandled error occurs executing the onSubscribe we will propagate it
            try {
                subscriber.onError(hook.onSubscribeError(e));
            } catch (OnErrorNotImplementedException e2) {
                // special handling when onError is not implemented ... we just rethrow
                throw e2;
            } catch (Throwable e2) {
                // if this happens it means the onError itself failed (perhaps an invalid function implementation)
                // so we are unable to propagate the error correctly and will just throw
                RuntimeException r = new RuntimeException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);
                // TODO could the hook be the cause of the error in the on error handling.
                hook.onSubscribeError(r);
                // TODO why aren't we throwing the hook's return value.
                throw r;
            }
            return Subscriptions.unsubscribed();
        }
    }

方法中hook.onSubsribeStart(this, onSubscribe).call(subscriber)默认情况下等价于onSubscribe.call(subscriber)。onSubscriber是什么呢？这个就需要了解BehaviorSubject的构造方法

protected BehaviorSubject(OnSubscribe<T> onSubscribe, SubjectSubscriptionManager<T> state) {
        super(onSubscribe);
        this.state = state;
    }

其中调用了父类Subject的构造方法

protected Subject(OnSubscribe<R> onSubscribe) {
        super(onSubscribe);
    }

其中调用了父类Observer的构造方法

protected Observable(OnSubscribe<T> f) {
        this.onSubscribe = f;
    }

onSubscribe即是BehaviorSubject构造方法中传入的第一个参数。

BehaviorSubject有3个静态工厂方法用来生产BehaviorSubject对象。

public final class BehaviorSubject<T> extends Subject<T, T> {

    public static <T> BehaviorSubject<T> create() {
        return create(null, false);
    }

    public static <T> BehaviorSubject<T> create(T defaultValue) {
        return create(defaultValue, true);
    }

    private static <T> BehaviorSubject<T> create(T defaultValue, boolean hasDefault) {
        final SubjectSubscriptionManager<T> state = new SubjectSubscriptionManager<T>();
        if (hasDefault) {
            state.set(NotificationLite.instance().next(defaultValue));
        }
        state.onAdded = new Action1<SubjectObserver<T>>() {

            @Override
            public void call(SubjectObserver<T> o) {
                o.emitFirst(state.get(), state.nl);
            }

        };
        state.onTerminated = state.onAdded;
        return new BehaviorSubject<T>(state, state); 
    }
    ....
}

前两个Public的静态构造方法实际上调用的是第三个private方法。

最后return new BehaviorSubject(state, state)，所以onSubscribe实际为一个SubjectSubscriptionManager的对象，onSubscribe.call(subscriber)实际调用的是SubjectSubscriptionManager的call方法。

/* package */final class SubjectSubscriptionManager<T> implements OnSubscribe<T> {
    ...
    @Override
    public void call(final Subscriber<? super T> child) {
        SubjectObserver<T> bo = new SubjectObserver<T>(child);
        addUnsubscriber(child, bo);
        onStart.call(bo);
        if (!child.isUnsubscribed()) {
            if (add(bo) && child.isUnsubscribed()) {
                remove(bo);
            }
        }
    }
}

• 调用addUnsubscriber方法，注册一个在取消订阅时执行的一个动作，即将该观擦者Observer移除掉。

/** Registers the unsubscribe action for the given subscriber. */
    void addUnsubscriber(Subscriber<? super T> child, final SubjectObserver<T> bo) {
        child.add(Subscriptions.create(new Action0() {
            @Override
            public void call() {
                remove(bo);
            }
        }));
    } 

• 调用add(SubjectObserver< T> o)方法，将该Observer加入已经注册的Observer[]数组当中。

boolean add(SubjectObserver<T> o) {
        do {
            State oldState = state;
            if (oldState.terminated) {
                onTerminated.call(o);
                return false;
            }
            State newState = oldState.add(o);
            if (STATE_UPDATER.compareAndSet(this, oldState, newState)) {
                onAdded.call(o);
                return true;
            }
        } while (true);
    }

该方法会调用onAdd.call(o)。BehaviorSubject的onAdd对象如下，state.get()得到的是最近的数据对象，o.emitFirst即会释放最近的数据对象，这正体现了BehaviorSubject的特点。

state.onAdded = new Action1<SubjectObserver<T>>() {

            @Override
            public void call(SubjectObserver<T> o) {
                o.emitFirst(state.get(), state.nl);
            }

        };

在这个过程中使用了SubjectSubscriptionManager的两个内部类。

1. State< T>
   该类用来管理已经注册的Observer数组，以及他们的状态。

/** State-machine representing the termination state and active SubjectObservers. */
    protected static final class State<T> {
        final boolean terminated;
        final SubjectObserver[] observers;
        static final SubjectObserver[] NO_OBSERVERS = new SubjectObserver[0];
        static final State TERMINATED = new State(true, NO_OBSERVERS);
        static final State EMPTY = new State(false, NO_OBSERVERS);

        public State(boolean terminated, SubjectObserver[] observers) {
            this.terminated = terminated;
            this.observers = observers;
        }
        public State add(SubjectObserver o) {
            ...
        }
        public State remove(SubjectObserver o) {
            ...
        }
    }

2.SubjectObserver
该类时Observer的一个装饰类，运用了装饰模式给Observer类添加新的功能。

以上就是Subject对象订阅Observer时的流程。

---

BehaviorSubject的onNext

Behavior的onNext(T v)方法如下

@Override
    public void onNext(T v) {
        Object last = state.get();
        if (last == null || state.active) {
            Object n = nl.next(v);
            for (SubjectObserver<T> bo : state.next(n)) {
                bo.emitNext(n, state.nl);
            }
        }
    }

state是SubjectSubscriptionManager类的对象，是这个对象来维护最近释放的数据对象，state.get()获取最近释放的数据对象，state.next(Object n)方法重新设置最近释放的数据对象，并返回已经注册的Observer数组。

SubjectObserver<T>[] next(Object n) {
        set(n);
        return state.observers;
    }

bo.emitNext(Object n, final NotificationLite< T> nl)释放给定的数据对象。

BehaviorSubject的onCompleted和onError

onCompleted和onError会调用SubjectSubscriptionManager的terminate(Object n)方法，该方法会重新设置最近释放的数据对象，设置Subject状态为TERMINATED，表示终结了，最后返回已注册的Observer数组。

SubjectObserver<T>[] terminate(Object n) {
        set(n);
        active = false;

        State<T> oldState = state;
        if (oldState.terminated) {
            return State.NO_OBSERVERS;
        }
        return STATE_UPDATER.getAndSet(this, State.TERMINATED).observers;
    }

参考文章：
http://www.liuhaihua.cn/archives/133614.html
http://blog.csdn.net/sun927/article/details/44818845