RxJava – 77博园

RxJava源码解析一

转载请标明出处，维权必究：http://77blogs.com/?p=512

本文基于RxJava 3.0.0

implementation 'io.reactivex.rxjava3:rxandroid:3.0.0'
implementation 'io.reactivex.rxjava3:rxjava:3.0.0'

1 2	implementation 'io.reactivex.rxjava3:rxandroid:3.0.0' implementation 'io.reactivex.rxjava3:rxjava:3.0.0'

我们从最简单的开始，挑最主要的讲：

Disposable disposable = Observable.create(new ObservableOnSubscribe<Object>() {
    @Override
    public void subscribe(ObservableEmitter<Object> e) throws Exception {
        e.onNext(1);
    }
}).subscribe(new Consumer<Object>() {
    @Override
    public void accept(Object o) throws Exception {
        Log.d("A", String.valueOf(o));
    }
});

Disposable disposable = Observable.create(new ObservableOnSubscribe<Object>() {

@Override

public void subscribe(ObservableEmitter<Object> e) throws Exception {

e.onNext(1);

}

}).subscribe(new Consumer<Object>() {

@Override

public void accept(Object o) throws Exception {

Log.d("A", String.valueOf(o));

}

});

我们先看create方法：

public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
    ObjectHelper.requireNonNull(source, "source is null");
    return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}

public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {

ObjectHelper.requireNonNull(source, "source is null");

return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));

}

该方法返回的是Observable对象，onAssembly方法可以占时不用理会，我们看

new ObservableCreate<T>(source)：

public ObservableCreate(ObservableOnSubscribe<T> source) {
    this.source = source;
}

public ObservableCreate(ObservableOnSubscribe<T> source) {

this.source = source;

}

也就是我们最终返回的是ObservableCreate对象，他继承Observable，持有ObservableOnSubscribe的引用。

接下来看subscribe方法：

public final Disposable subscribe(Consumer<? super T> onNext) {
    return subscribe(onNext, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION, Functions.emptyConsumer());
}

public final Disposable subscribe(Consumer<? super T> onNext) {

return subscribe(onNext, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION, Functions.emptyConsumer());

}

接着：

public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
        Action onComplete, Consumer<? super Disposable> onSubscribe) {
    ObjectHelper.requireNonNull(onNext, "onNext is null");
    ObjectHelper.requireNonNull(onError, "onError is null");
    ObjectHelper.requireNonNull(onComplete, "onComplete is null");
    ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");
    LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);
    subscribe(ls);
    return ls;
}

public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,

Action onComplete, Consumer<? super Disposable> onSubscribe) {

ObjectHelper.requireNonNull(onNext, "onNext is null");

ObjectHelper.requireNonNull(onError, "onError is null");

ObjectHelper.requireNonNull(onComplete, "onComplete is null");

ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");

LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);

subscribe(ls);

return ls;

}

可以看出，我们传进来的Consumer对象被放进了LambdaObserver中。接下来重点是subscribe(ls);

public final void subscribe(Observer<? super T> observer) {
    ObjectHelper.requireNonNull(observer, "observer is null");
    try {
        observer = RxJavaPlugins.onSubscribe(this, observer);
        ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");
        subscribeActual(observer);
    } catch (NullPointerException e) { // NOPMD
        throw e;
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        // can't call onError because no way to know if a Disposable has been set or not
        // can't call onSubscribe because the call might have set a Subscription already
        RxJavaPlugins.onError(e);
        NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
        npe.initCause(e);
        throw npe;
    }
}

public final void subscribe(Observer<? super T> observer) {

ObjectHelper.requireNonNull(observer, "observer is null");

try {

observer = RxJavaPlugins.onSubscribe(this, observer);

ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");

subscribeActual(observer);

} catch (NullPointerException e) { // NOPMD

throw e;

} catch (Throwable e) {

Exceptions.throwIfFatal(e);

// can't call onError because no way to know if a Disposable has been set or not

// can't call onSubscribe because the call might have set a Subscription already

RxJavaPlugins.onError(e);

NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");

npe.initCause(e);

throw npe;

}

subscribeActual(observer);是一个抽象方法，实现它的方法有很多，从上面的分析中可以知道，调用下面方法的是ObservableCreate对象。

.subscribe(new Consumer<Object>() {
    @Override
    public void accept(Object o) throws Exception {
        Log.d("A", String.valueOf(o));
    }
});

.subscribe(new Consumer<Object>() {

@Override

public void accept(Object o) throws Exception {

Log.d("A", String.valueOf(o));

}

});

所以subscribeActual(observer);调用的是ObservableCreate里面的。

@Override
protected void subscribeActual(Observer<? super T> observer) {
    CreateEmitter<T> parent = new CreateEmitter<T>(observer);
    observer.onSubscribe(parent);
    try {
        source.subscribe(parent);
    } catch (Throwable ex) {
        Exceptions.throwIfFatal(ex);
        parent.onError(ex);
    }
}

@Override

protected void subscribeActual(Observer<? super T> observer) {

CreateEmitter<T> parent = new CreateEmitter<T>(observer);

observer.onSubscribe(parent);

try {

source.subscribe(parent);

} catch (Throwable ex) {

Exceptions.throwIfFatal(ex);

parent.onError(ex);

}

这个方法里面构造了一个发射器CreateEmitter，传进去的observer就是上面所说的LambdaObserver，里面持有我们传进去的Consumer对象。

observer.onSubscribe(parent);调用的是LambdaObserver里面的onSubscribe；

@Override
public void onSubscribe(Disposable s) {
    if (DisposableHelper.setOnce(this, s)) {
        try {
            onSubscribe.accept(this);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            onError(ex);
        }
    }

@Override

public void onSubscribe(Disposable s) {

if (DisposableHelper.setOnce(this, s)) {

try {

onSubscribe.accept(this);

} catch (Throwable ex) {

Exceptions.throwIfFatal(ex);

onError(ex);

}

这个方法里面的onSubscribe.accept(this);中的onSubscribe，是我们上面的方法中的第四个参数，是一个空的Consumer，所以不用理他：

public final Disposable subscribe(Consumer<? super T> onNext) {
    return subscribe(onNext, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION, Functions.emptyConsumer());
}

public final Disposable subscribe(Consumer<? super T> onNext) {

return subscribe(onNext, Functions.ERROR_CONSUMER, Functions.EMPTY_ACTION, Functions.emptyConsumer());

}

我们回过头看subscribeActual方法中的source.subscribe(parent);

source是我们之前传进去的ObservableOnSubscribe对象，source.subscribe便是调用ObservableOnSubscribe的subscribe方法，传进去的是一个发射器。

到此我们总算调用到了外面这个方法了。

（也就是说必须调用subscribe方法，才能发射事件）

Disposable disposable = Observable.create(new ObservableOnSubscribe<Object>() {
    @Override
    public void subscribe(ObservableEmitter<Object> e) throws Exception {
        e.onNext(1);
    }
})

Disposable disposable = Observable.create(new ObservableOnSubscribe<Object>() {

@Override

public void subscribe(ObservableEmitter<Object> e) throws Exception {

e.onNext(1);

}

})

我们再看e.onNext(1)，在ObservableCreate的内部类CreateEmitter里面：

@Override
public void onNext(T t) {
    if (t == null) {
        onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
        return;
    }
    if (!isDisposed()) {
        observer.onNext(t);
    }
}

@Override

public void onNext(T t) {

if (t == null) {

onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));

return;

}

if (!isDisposed()) {

observer.onNext(t);

}

这里的observer我们上面已经分析过，是构造发射器的时候传进去的LambdaObserver，LambdaObserver里面持有我们传进去的Consumer对象。

于是observer.onNext(t)便是调用LambdaObserver的onNext(t)

@Override
public void onNext(T t) {
    if (!isDisposed()) {
        try {
            onNext.accept(t);
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            onError(e);
        }
    }
}

@Override

public void onNext(T t) {

if (!isDisposed()) {

try {

onNext.accept(t);

} catch (Throwable e) {

Exceptions.throwIfFatal(e);

onError(e);

}

onNext便是它持有的的Consumer对象，于是onNext.accept(t)便回调到了外面这一层的accept：

Disposable disposable = Observable.create(new ObservableOnSubscribe<Object>() {
    @Override
    public void subscribe(ObservableEmitter<Object> e) throws Exception {
        e.onNext(1);
    }
}).subscribe(new Consumer<Object>() {
    @Override
    public void accept(Object o) throws Exception {
        Log.d("A", String.valueOf(o));
    }
});

Disposable disposable = Observable.create(new ObservableOnSubscribe<Object>() {

@Override

public void subscribe(ObservableEmitter<Object> e) throws Exception {

e.onNext(1);

}

}).subscribe(new Consumer<Object>() {

@Override

public void accept(Object o) throws Exception {

Log.d("A", String.valueOf(o));

}

});

到此结束。

RxJava基本使用

转载请标明出处：http://77blogs.com/?p=162

RxJava究竟是啥，从根本上来讲，它就是一个实现异步操作的库，并且能够使代码非常简洁。它的异步是使用观察者模式来实现的。

关于观察者模式的介绍，可以看我这一篇文章：

https://www.cnblogs.com/tangZH/p/11175120.html

这里我主要讲RxJava的一些基本用法，基本案例，原理的话暂时不深究：

一、自己构造事件

Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter emitter) {
                int i = getNumber();
                if (i < 0) {
                    emitter.onComplete();
                    return;
                } else {
                    Log.d(TAG, Thread.currentThread().getName());
                    emitter.onNext(i);
                    emitter.onComplete();
                }
            }
        })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, Thread.currentThread().getName());
                        Log.d(TAG, integer + "");
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {

                    }
                });

Observable.create(new ObservableOnSubscribe<Integer>() {

@Override

public void subscribe(ObservableEmitter emitter) {

int i = getNumber();

if (i < 0) {

emitter.onComplete();

return;

} else {

Log.d(TAG, Thread.currentThread().getName());

emitter.onNext(i);

emitter.onComplete();

}

})

.subscribeOn(Schedulers.io())

.observeOn(AndroidSchedulers.mainThread())

.subscribe(new Consumer<Integer>() {

@Override

public void accept(Integer integer) throws Exception {

Log.d(TAG, Thread.currentThread().getName());

Log.d(TAG, integer + "");

}

}, new Consumer<Throwable>() {

@Override

public void accept(Throwable throwable) throws Exception {

}

});

RxJava 有四个基本概念：Observable (可观察者，即被观察者)、 Observer (观察者)、 subscribe (订阅)、事件。Observable 和 Observer 通过 subscribe() 方法实现订阅关系，从而 Observable 可以在需要的时候发出事件来通知 Observer。

onNext():方法用来发送事件。

下面看看其他两个方法：

onCompleted(): 事件队列完结。RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。RxJava 规定，当不会再有新的 onNext() 发出时，需要触发 onCompleted() 方法作为标志。
onError(): 事件队列异常。在事件处理过程中出异常时，onError() 会被触发，同时队列自动终止，不允许再有事件发出。
在一个正确运行的事件序列中, onCompleted() 和 onError() 有且只有一个，并且是事件序列中的最后一个。需要注意的是，onCompleted() 和 onError() 二者也是互斥的，即在队列中调用了其中一个，就不应该再调用另一个。

讲一下我们上面的例子，上面这个例子是采用简洁的链式调用来写的：

首先使用 create() 方法来创建一个 Observable ，并为它定义事件触发规则，然后通过emitter.onNext(i)传递出来，.subscribeOn(Schedulers.io())便是指定该事件产生的所在的线程为子线程，.observeOn(AndroidSchedulers.mainThread())指定观察者执行的线程为主线程。这时候为止返回的对象为Observable对象。

然后该Observable对象subscribe绑定观察者（也就是观察者进行订阅），里面有接收被观察者发出来的事件，有一个成功的方法，和一个失败的方法，这样就实现了由被观察者向观察传递事件。

二、对集合里的数据进行变换

        List<Integer> list = new ArrayList<Integer>() {
            {
                add(0);
                add(1);
                add(2);
            }
        };
        Observable.fromIterable(list).map(new Function() {
            @Override
            public Object apply(Object o) throws Exception {
                int i = (int) o + 1;
                return String.valueOf(i);
            }
        })
                .toList()
                .toObservable().subscribeOn(Schedulers.io())
                .subscribeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer() {
                    @Override
                    public void accept(Object o) throws Exception {
                        Log.d(TAG, o.toString());
                    }
                });

List<Integer> list = new ArrayList<Integer>() {

{

add(0);

add(1);

add(2);

}

};

Observable.fromIterable(list).map(new Function() {

@Override

public Object apply(Object o) throws Exception {

int i = (int) o + 1;

return String.valueOf(i);

}

})

.toList()

.toObservable().subscribeOn(Schedulers.io())

.subscribeOn(AndroidSchedulers.mainThread())

.subscribe(new Consumer() {

@Override

public void accept(Object o) throws Exception {

Log.d(TAG, o.toString());

}

});

且看，我们需要对某个集合里面的数据一一进行变换，然后发送出来执行其他操作。

上面便是对集合里面的每一项进行加一操作，然后再转换为String类型，然后toList()，组合成集合发送出来，最后在观察者方法中打印出每一项。

三、合并执行

定义两个被观察者，各自产生事件，然后合并在一起，发送给一个观察者。

首先定义我们上面第一个例子的被观察者，用于发送一个数字：

        Observable observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter emitter) {
                int i = getNumber();
                if (i < 0) {
                    emitter.onComplete();
                    return;
                } else {
                    Log.d(TAG, Thread.currentThread().getName());
                    emitter.onNext(i);
                    emitter.onComplete();
                }
            }
        })
                .subscribeOn(Schedulers.io());

Observable observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {

@Override

public void subscribe(ObservableEmitter emitter) {

int i = getNumber();

if (i < 0) {

emitter.onComplete();

return;

} else {

Log.d(TAG, Thread.currentThread().getName());

emitter.onNext(i);

emitter.onComplete();

}

})

.subscribeOn(Schedulers.io());

其次再定义我们上面第二个例子的被观察者：

        List<Integer> list = new ArrayList<Integer>() {
            {
                add(0);
                add(1);
                add(2);
            }
        };
        Observable observable2 = Observable.fromIterable(list).map(new Function() {
            @Override
            public Object apply(Object o) {
                int i = (int) o + 1;
                return String.valueOf(i);
            }
        })
                .toList()
                .toObservable().subscribeOn(Schedulers.io());

List<Integer> list = new ArrayList<Integer>() {

{

add(0);

add(1);

add(2);

}

};

Observable observable2 = Observable.fromIterable(list).map(new Function() {

@Override

public Object apply(Object o) {

int i = (int) o + 1;

return String.valueOf(i);

}

})

.toList()

.toObservable().subscribeOn(Schedulers.io());

最后将这两个被观察者的事件合并起来发送给一个观察者：

        Disposable disposable = Observable.zip(observable1, observable2, new BiFunction() {
            @Override
            public Object apply(Object o, Object o2) throws Exception {
                int i = (int) o;
                String k = (String) ((List) o2).get(0);
                return k + i;
            }
        })
                .subscribe(new Consumer() {
                    @Override
                    public void accept(Object o) {
                        Log.d(TAG, (String) o);
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) {
                        Log.d(TAG, throwable.getMessage());
                    }
                });

Disposable disposable = Observable.zip(observable1, observable2, new BiFunction() {

@Override

public Object apply(Object o, Object o2) throws Exception {

int i = (int) o;

String k = (String) ((List) o2).get(0);

return k + i;

}

})

.subscribe(new Consumer() {

@Override

public void accept(Object o) {

Log.d(TAG, (String) o);

}

}, new Consumer<Throwable>() {

@Override

public void accept(Throwable throwable) {

Log.d(TAG, throwable.getMessage());

}

});

zip方法，顾名思义，有点类似与于打包的意思。

o为被观察者1返回的结果，o2为被观察2返回的结果，将这两个结果一起处理后发送给观察者。打印出来。

现在先介绍这几个，找个时间再整理一些其他的用法以及原理实现。