问在rxjava和rxandroid中的可观察的正确解释

EN

public class CollectionObservable<T> extends Observable<T> {

    private Collection<T> collection;
    private List<Observer<? super T>> observers;

    public CollectionObservable(Collection<T> collection) {
        if (collection != null) {
            this.collection = collection;
        }
        this.observers = new ArrayList<>(2);
    }

    public Collection<T> getCollection() {
        return collection;
    }

    public void setCollection(Collection<T> collection) {
        this.collection = collection;
        emitValuesToAllObserver();
    }

    public void complete() {
        if (this.collection != null) {
            for (Observer<? super T> observer : this.observers) {
                observer.onComplete();
            }
        }
    }

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        this.observers.add(observer);
        emitValues(observer);
    }

    private void emitValuesToAllObserver() {
        for (Observer<? super T> observer : this.observers) {
            emitValues(observer);
        }
    }

    private void emitValues(Observer<? super T> observer) {
        if (this.collection != null) {
            for (T obj : this.collection) {
                observer.onNext(obj);
            }
        }
    }
}