Rxjava詳解

一,作用

RxJava的目的就是異步。RxJava的特點就是可以非常簡便的實現(xiàn)異步調(diào)用，可以在邏輯復(fù)雜的代碼邏輯中以比較輕易的方式實現(xiàn)異步調(diào)用。隨著邏輯的復(fù)雜，需求的更改，代碼可依然能保持極強(qiáng)的閱讀性，在深入的使用過程中一定對這點深有體會。

二,添加依賴

io.reactivex.rxjava2:rxjava:2.0.4

io.reactivex.rxjava2:rxjava:2.0.4

三,關(guān)鍵api

Observable：在觀察者模式中稱為“被觀察者”； Observer：觀察者模式中的“觀察者”，可接收Observable發(fā)送的數(shù)據(jù)； subscribe：訂閱，觀察者與被觀察者，通過subscribe()方法進(jìn)行訂閱； Subscriber：也是一種觀察者，在2.0中 它與Observer沒什么實質(zhì)的區(qū)別，不同的是 Subscriber要與Flowable(也是一種被觀察者)聯(lián)合使用，該部分內(nèi)容是2.0新增的，后續(xù)文章再介紹。Obsesrver用于訂閱Observable，而Subscriber用于訂閱Flowable

四,回調(diào)方法

onComplete(): 事件隊列完結(jié)時調(diào)用該方法。RxJava 不僅把每個事件單獨處理，還會把它們看做一個隊列。 onError(): 事件隊列異常。在事件處理過程中出異常時，onError() 會被觸發(fā)，同時隊列自動終止，不允許再有事件發(fā)出。 onSubscribe()：RxJava 2.0 中新增的，傳遞參數(shù)為Disposable ，Disposable 相當(dāng)于RxJava1.x中的Subscription,用于解除訂閱。 注意：onComplete() 和 onError() 二者也是互斥的，即在隊列中調(diào)用了其中一個，就不應(yīng)該再調(diào)用另一個。

五,線程調(diào)度,****Scheduler，**相當(dāng)于線程控制器，RxJava 通過它來指定每一段代碼應(yīng)該運行在什么樣的線程。

1,Schedulers.immediate():** 直接在當(dāng)前線程運行，相當(dāng)于不指定線程。這是默認(rèn)的 Scheduler。

2,Schedulers.newThread():** 總是啟用新線程，并在新線程執(zhí)行操作

3,Schedulers.io():** I/O 操作（讀寫文件、讀寫數(shù)據(jù)庫、網(wǎng)絡(luò)信息交互等）所使用的 Scheduler,多數(shù)情況下 io() 比 newThread() 更有效率

4,Schedulers.computation()**: **計算所使用的 Scheduler,不要把 I/O 操作放在 computation() 中，否則 I/O 操作的等待時間會浪費 CPU。

5,Android 還有一個專用的** **AndroidSchedulers.mainThread()****，它指定的操作將在 Android 主線程運行。

subscribeOn(): 指定Observable(被觀察者)所在的線程，或者叫做事件產(chǎn)生的線程。 * observeOn(): 指定 Observer(觀察者)所運行在的線程，或者叫做事件消費的線程。

Disposable*, 這個單詞的字面意思是一次性用品,用完即可丟棄的。在RxJava中,用它來切斷Observer(觀察者)與Observable(被觀察者)之間的連接，當(dāng)調(diào)用它的dispose()方法時, 它就會將Observer(觀察者)與Observable(被觀察者)之間的連接切斷, 從而導(dǎo)致Observer(觀察者)收不到事件。

六,代碼
///第二種方式,采用鏈?zhǔn)秸{(diào)用/
private void second() {
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("hello world");
e.onNext("第一次用");
e.onComplete();
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.e(TAG, "onSubscribe: ");
}

        @Override
        public void onNext(String s) {
            Log.e(TAG, s);
        }

        @Override
        public void onError(Throwable e) {
            Log.e(TAG, e.getMessage());
        }

        @Override
        public void onComplete() {
            Log.e(TAG, "onComplete");
        }
    });
}

/*
第三種方式,使用Consumer只接收發(fā)送信息/
private void third() {
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("hello world");
e.onNext("第一次用");
e.onComplete();
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
//表示我只在乎next發(fā)出的消息,也只接受這個消息
Log.e(TAG, s);
}
});
}

///第四種,調(diào)用Schedulers線程調(diào)度器/
private void four() {
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
Log.e(TAG, "事件產(chǎn)生線程" + Thread.currentThread().getName());
e.onNext("hehe");

        }
    }).subscribeOn(Schedulers.io()) //事件產(chǎn)生的線程,也就是被觀察者
            .observeOn(AndroidSchedulers.mainThread()) //時間消費的線程,也就是觀察者
            .subscribe(new Consumer<String>() {
                @Override
                public void accept(@NonNull String s) throws Exception {
                    Log.e(TAG, "接受到的結(jié)果" + s);
                    Log.e(TAG, "事件消費線程" + Thread.currentThread().getName());
                }
            });
}

/*
第五種,使用just直接出發(fā)onNext()
* 使用just( )，將為你創(chuàng)建一個Observable并自動為你調(diào)用onNext( )發(fā)射數(shù)據(jù)。
* 通過just( )方式 直接觸發(fā)onNext()，just中傳遞的參數(shù)將直接在Observer的onNext()方法中接收到。*/
private void five() {
Observable.just("hello").subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.e(TAG, s);
}
});
}

///第六種,使用interval做定時器,/
private void six() {
//可用作定時器。即按照固定2秒一次調(diào)用onNext()方法。
Observable<Long> observable = Observable.interval(2, TimeUnit.SECONDS);
Observer<Long> observer = new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
}

        @Override
        public void onNext(Long aLong) {
            Log.e(TAG, aLong.toString());
        }

        @Override
        public void onError(Throwable e) {
        }

        @Override
        public void onComplete() {

        }
    };
    SystemClock.sleep(6000);//睡眠6秒后，才進(jìn)行訂閱, 仍然從0開始，表示Observable內(nèi)部邏輯剛開始執(zhí)行
    observable.subscribe(observer);
}

/*
* //第七種,使用map將一種數(shù)據(jù)轉(zhuǎn)換成另一種數(shù)據(jù)返回*/
private void seven() {
Observable.just("hello").map(new Function<String, Integer>() {
@Override
public Integer apply(String s) throws Exception {
return s.length();
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(@NonNull Integer integer) throws Exception {
Log.e(TAG, "返回的數(shù)據(jù) " + integer);
}
});
}

/*
第八種,使用filter將需要的數(shù)據(jù)篩選出來
* */
private void eight() {
lists.add("你好呵呵呵呵呵");
lists.add("開心哈哈哈哈");
Observable.just(lists).flatMap(new Function<List<String>, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(List<String> strings) throws Exception {
return Observable.fromIterable(strings);
}
}).filter(new Predicate<Object>() {
@Override
public boolean test(Object s) throws Exception {
String newStr = (String) s;
if (newStr.charAt(5) - '0' > 5) {
return true;
}
return false;
}
}).subscribe(new Consumer<Object>() {
@Override
public void accept(Object o) throws Exception {
System.out.println((String) o);
Log.e(TAG, "返回的數(shù)據(jù) " + (String) o);
}
});
}

/*
* 第九種,使用Flowable處理背壓問題(所謂背壓，即生產(chǎn)者的速度大于消費者的速度帶來的問題。)
* */
private void nine() {
Flowable<Integer> flowable = Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
Log.e(TAG, "我是1");
emitter.onNext(1);
Log.e(TAG, "我是 2");
emitter.onNext(2);
Log.e(TAG, "我是 3");
emitter.onNext(3);
Log.e(TAG, "我是 complete");
emitter.onComplete();
}
}, BackpressureStrategy.ERROR); //增加了一個參數(shù)

    Subscriber<Integer> subscriber = new Subscriber<Integer>() {
        @Override
        public void onSubscribe(Subscription s) {
            Log.e(TAG, "onSubscribe");
            s.request(Long.MAX_VALUE);
        }

        @Override
        public void onNext(Integer integer) {
            Log.e(TAG, "onNext: " + integer);
        }

        @Override
        public void onError(Throwable t) {
            Log.e(TAG, "onError: ", t);
        }

        @Override
        public void onComplete() {
            Log.e(TAG, "onComplete");
        }
    };
    flowable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(subscriber);
}

/*
//第十種點擊“開始”按鈕，建立連接。生產(chǎn)者開始生產(chǎn)事件，剛開始消費者通過request()只要了50個事件消費。
// 然后每次點擊“消費”按鈕，再次消費50個事件。

   * 可以看出，生產(chǎn)者一次性傳入128個事件進(jìn)入緩存池。點擊“開始”按鈕，消費了50個。
   * 然后第一次點擊“消費”按鈕，又消費了50個，第二次點擊“消費”按鈕，再次消費50個。
   * 然而此時原來的128個緩存只剩下28個了，所以先消費掉28個，
   * 然后剩下22個是后來傳入的（其實后來的是在消費了96個后傳入，并一次性在緩存池中又傳入了96個）。*/
private void ten() {
    mFlowable = Flowable.create(new FlowableOnSubscribe<Integer>() {
        @Override
        public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
            for (int i = 0;i< 300 ; i++) {
                emitter.onNext(i);
            }
        }
    }, BackpressureStrategy.DROP);

    mSubscriber = new Subscriber<Integer>() {
        @Override
        public void onSubscribe(Subscription s) {
            mSubscription = s;
            s.request(50);
        }

        @Override
        public void onNext(Integer integer) {
            Log.e(TAG, "onNext: " + integer);
        }

        @Override
        public void onError(Throwable t) {
            Log.e(TAG, "onError: "+t);
        }

        @Override
        public void onComplete() {
            Log.e(TAG, "onComplete: ");
        }
    };
}

public void consume() {
    mFlowable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(mSubscriber);
}

public void consumeAgain() {
    mSubscription.request(50);
}

@Override
public void onClick(View view) {
    switch (view.getId()) {
        case R.id.consume:
            consume();
            break;
        case R.id.consume_again:
            consumeAgain();
            break;
    }
}

借鑒<u>Ruheng</u>的博客

<u>http://www.itdecent.cn/p/310726a75045</u>

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容