RxJava简析-开源中国

rxjava文档地址https://mcxiaoke.gitbooks.io/rxdocs/content/ 这个是中文版的

android studio 添加依赖 implementation 'io.reactivex.rxjava3:rxjava:3.0.4'

首先，打印helloworld：

public void hello(String args){    Flowable.fromArray(args).subscribe(s -> System.out.println("hello " + s + "!"));}

跟以前其他语言不大一样，看上去很麻烦，我们一步步来看

Flowable.fromArray(args)

这个方法最重要的就是里面的最后一句

new FlowableFromArray(items)

果然FlowableFromArray是Flowable的子类，所以真正的实现在子类里面

Flowable.fromArray(args).subscribe

subscribe进到里面的是

public final Disposable subscribe(@NonNull Consumer super T> onNext, @NonNull Consumer super Throwable> onError,        @NonNull Action onComplete) {    Objects.requireNonNull(onNext, "onNext is null");    Objects.requireNonNull(onError, "onError is null");    Objects.requireNonNull(onComplete, "onComplete is null");
    LambdaSubscriber ls = new LambdaSubscriber(onNext, onError, onComplete, FlowableInternalHelper.RequestMax.INSTANCE);
    subscribe(ls);
    return ls;}

看上去最重要的就是这两句了

LambdaSubscriberT> ls = new LambdaSubscriber(onNext, onError, onComplete, FlowableInternalHelper.RequestMax.INSTANCE);
subscribe(ls);

先进到subscribe(ls)中，发现这句

subscribeActual(flowableSubscriber)

跳进去发现是个抽象方法，那么实现肯定在子类啦，进到子类FlowableFromArray

@Overridepublic void subscribeActual(Subscriber super T> s) {    if (s instanceof ConditionalSubscriber) {        s.onSubscribe(new ArrayConditionalSubscription(                (ConditionalSubscriber super T>)s, array));    } else {        s.onSubscribe(new ArraySubscription(s, array));    }}

跳进去又发现onSubscribe是个抽象方法，那么实现方法在哪呢，对啦，就是之前看到的LambdaSubscriber

public void onSubscribe(Subscription s) {    if (SubscriptionHelper.setOnce(this, s)) {        try {            onSubscribe.accept(this);        } catch (Throwable ex) {            Exceptions.throwIfFatal(ex);            s.cancel();            onError(ex);        }    }}

这个onSubscribe.accept(this)跳过去就是接口Consumer的accept方法了

所以一开始的helloworld代码也可以改成

FlowableFromArray flowableFromArray = new FlowableFromArray(new String[]{args});flowableFromArray.subscribe(new ConsumerString>() {    @Override    public void accept(String s) throws Throwable {        System.out.println("hello " + s + "!");    }});

是不是很麻烦，饶了一大圈，没关系，我们继续往下看

这里给出一些名词的翻译

Reactive 直译为反应性的，有活性的，根据上下文一般翻译为反应式、响应式
Iterable 可迭代对象，支持以迭代器的形式遍历，许多语言中都存在这个概念
Observable 可观察对象，在Rx中定义为更强大的Iterable，在观察者模式中是被观察的对象，一旦数据产生或发生变化，会通过某种方式通知观察者或订阅者
Observer 观察者对象，监听Observable发射的数据并做出响应，Subscriber是它的一个特殊实现
emit 直译为发射，发布，发出，含义是Observable在数据产生或变化时发送通知给Observer，调用Observer对应的方法，文章里一律译为发射
items 直译为项目，条目，在Rx里是指Observable发射的数据项，文章里一律译为数据，数据项

下面是常用的操作符列表：

创建操作 Create, Defer, Empty/Never/Throw, From, Interval, Just, Range, Repeat, Start, Timer
变换操作 Buffer, FlatMap, GroupBy, Map, Scan和Window
过滤操作 Debounce, Distinct, ElementAt, Filter, First, IgnoreElements, Last, Sample, Skip, SkipLast, Take, TakeLast
组合操作 And/Then/When, CombineLatest, Join, Merge, StartWith, Switch, Zip
错误处理 Catch和Retry
辅助操作 Delay, Do, Materialize/Dematerialize, ObserveOn, Serialize, Subscribe, SubscribeOn, TimeInterval, Timeout, Timestamp, Using
条件和布尔操作 All, Amb, Contains, DefaultIfEmpty, SequenceEqual, SkipUntil, SkipWhile, TakeUntil, TakeWhile
算术和集合操作 Average, Concat, Count, Max, Min, Reduce, Sum
转换操作 To
连接操作 Connect, Publish, RefCount, Replay
反压操作，用于增加特殊的流程控制策略的操作符

下面我们来看第一个操作符：Create

Observable.create(new Observable.OnSubscribe() {    @Override    public void call(Subscriber super Integer> observer) {        try {            if (!observer.isUnsubscribed()) {                for (int i = 1; i 5; i++) {                    observer.onNext(i);                }                observer.onCompleted();            }        } catch (Exception e) {            observer.onError(e);        }    } } ).subscribe(new Subscriber() {        @Override        public void onNext(Integer item) {            System.out.println("Next: " + item);        }
        @Override        public void onError(Throwable error) {            System.err.println("Error: " + error.getMessage());        }
        @Override        public void onCompleted() {            System.out.println("Sequence complete.");        }    });

我们一起来看源码

首先是Observable的create方法

public final static  Observable create(OnSubscribe f) {    return new Observable(hook.onCreate(f));}

这里没什么，就是返回创建一个Observable对象，但是要注意里面的参数OnSubscribe

public interface OnSubscribeT> extends Action1Subscriber super T>> {    // cover for generics insanity}
public interface Action1T> extends Action {    void call(T t);}

这个参数是一个接口，它的父类里有个抽象待实现的方法call，而且call方法被传了Subscriber进去

我们来看Subscriber这个类，原来是个接口，而且它的父类Observer有三个很重要的方法

public interface ObserverT> {    void onCompleted();    void onError(Throwable e);    void onNext(T t);}

第一个create方法算是完成了，我们可以拆分来看

Observable integerObservable = Observable.create(new Observable.OnSubscribe() {    @Override    public void call(Subscriber super Integer> observer) {        try {            if (!observer.isUnsubscribed()) {                for (int i = 0; i 5; i++) {                    observer.onNext(i);                }                observer.onCompleted();            }        } catch (Exception e) {            observer.onError(e);        }    }});

第二个方法subscribe，它的参数也是Subscriber,即intergerObservable.subscribe(Subscriber)

所以我们就看出来了，Observable这个被观察者先是通过call增加一系列的监听，然后通过subscribe订阅监听。这样，当call里的内容开始执行后，触发监听回调

下面我要放大招了，我把源码简化了一下

public interface MyOnSubscribe {    void call(MySubscriber subscriber);}
public interface MySubscriber {    void onNext();
    void onCompleted();
    void onError();}
public class MyObservable {
    MyOnSubscribe onSubscribe;
    public MyObservable(MyOnSubscribe onSubscribe) {        this.onSubscribe = onSubscribe;    }
    public final static MyObservable create(MyOnSubscribe onSubscribe) {        return new MyObservable(onSubscribe);    }
    public final void subscribe(MySubscriber subscriber) {        onSubscribe.call(subscriber);    }}

测试代码

public void hello() {        MyObservable.create(new MyOnSubscribe() {            @Override            public void call(MySubscriber subscriber) {                try {                    for (int i = 0; i 5; i++) {                        subscriber.onNext();                    }                    subscriber.onCompleted();                } catch (Exception e) {                    subscriber.onError();                }            }        }).subscribe(new MySubscriber() {            @Override            public void onNext() {                System.out.println(1);            }
            @Override            public void onCompleted() {                System.out.println("onCompleted");            }
            @Override            public void onError() {                System.out.println("onError");            }        });    }

得到的结果是一样的。所以说，代码万变不离其中，只要灵活运用接口，接口就是用来监听的

第二个操作符from

Integer[] items = {0, 1, 2, 3, 4, 5};Observable.from(items).subscribe(new Action1() {    @Override    public void call(Integer integer) {        System.out.println(integer);    }});

先看Observable的from方法

public final static  Observable from(T[] array) {    return from(Arrays.asList(array));}

其实就是把数组转成list，但是再点from进去就很重要

public final static  Observable from(Iterable extends T> iterable) {    return create(new OnSubscribeFromIterable(iterable));}public OnSubscribeFromIterable(Iterable extends T> iterable) {    if (iterable == null) {        throw new NullPointerException("iterable must not be null");    }    this.is = iterable;}

OnSubscribeFromIterable是继承自OnSubscribe的，所以后面调的call方法，实际上是调的OnSubscribeFromIterable里的call方法，我们来看一下

@Overridepublic void call(final Subscriber super T> o) {    final Iterator extends T> it = is.iterator();    if (!it.hasNext() && !o.isUnsubscribed())        o.onCompleted();    else         o.setProducer(new IterableProducer(o, it));}

真相大白了，在这里做了迭代。还有一个操作符just，其实底层里面调的就是from，只不过还限制了参数个数，而且参数类型必须相同，感觉用处不大

第三个操作符repeat

Observable.just(1, 2).repeat(4).subscribe(new Action1() {    @Override    public void call(Integer integer) {        System.out.println(integer);    }});

repeat点进去是OnSubcribRedo.repeat，紧追着count这个参数，会看到一个RedoFinite类

public static final class RedoFinite implements Func1Observable extends Notification>>, Observable>> {    private final long count;
    public RedoFinite(long count) {        this.count = count;    }
    @Override    public Observable> call(Observable extends Notification>> ts) {        return ts.map(new Func1, Notification>>() {
            int num=0;                        @Override            public Notification> call(Notification> terminalNotification) {                if(count == 0) {                    return terminalNotification;                }                                num++;                if(num                     return Notification.createOnNext(num);                } else {                    return terminalNotification;                }            }                    }).dematerialize();    }}

这里就看到了，有个num++和num

第4个操作符Map和flapMap

这两个变换操作符可谓非常重要，经常用到，我写了4个例子，请仔细区别，就可以知道它们到底做了什么

Student student1 = new Student("stark", new Course[]{new Course("Chinese"), new Course("English")});Student student2 = new Student("adam", new Course[]{new Course("Math"), new Course("Physical")});
Student[] students = new Student[]{student1, student2};Observable.from(students).subscribe(new Action1() {    @Override    public void call(Student student) {        System.out.println(student.getName());    }});
System.out.println("-------------");
Observable.from(students).map(new Func1() {    @Override    public String call(Student student) {        return student.getName();    }}).subscribe(new Action1() {    @Override    public void call(String name) {        System.out.println(name);    }});
System.out.println("-------------");
Observable.from(students).map(new Func1() {    @Override    public Course[] call(Student student) {        return student.getCourses();    }}).subscribe(new Action1() {    @Override    public void call(Course[] courses) {        System.out.println(courses[0].getName());        System.out.println(courses[1].getName());    }});
System.out.println("-------------");
Observable.from(students).flatMap(new Func1>() {    @Override    public Observable call(Student student) {        return Observable.from(student.getCourses());    }}).subscribe(new Action1() {    @Override    public void call(Course course) {        System.out.println(course.getName());    }});
输出：starkadam-------------starkadam-------------ChineseEnglishMathPhysical-------------ChineseEnglishMathPhysical

如果你仔细看代码，就会发现map就是一对一的转换，flatMap是一对多的转换，转换的前后类型在方法Func1中已经标的很清楚。例子：Func1(Student,String)就代表传参是Student，返回类型是String，具体的实现在call里面student.getName()

map和flatMap可以看作是将我们经常用到的嵌套循环for(i){for(j)...}...给解耦了，看起来更清楚一些，中间可以插入更多的操作

源码里面的实现就是迭代，没什么好说

第5个操作符filter：

Observable.just(1,2,3,4,5).filter(new Func1Boolean>() {    @Override    public Boolean call(Integer integer) {        return integer4;    }}).subscribe(new Action1() {    @Override    public void call(Integer integer) {        System.out.println(integer);    }});

先过滤再循环输出

第6个组合操作符and/then/when

implementation 'io.reactivex:rxjava-joins:0.22.0'
ObservableString> just1 = Observable.just("A", "B");Observable just2 = Observable.just(1, 2, 3);Pattern2String, Integer> pattern = JoinObservable.from(just1).and(just2);Plan0String> plan = pattern.then(new Func2

RxJava简析

参与评论

评论留言

还没有评论留言，赶紧来抢楼吧~~

给这篇文章打个标签吧~