日本激情小视频,国产日韩伦理精AV

參考：
http://blog.csdn.net/maplejaw_/article/details/52442065
http://www.itdecent.cn/nb/5864063

去年RxJava2.x發(fā)布了，與RxJava1.x相比，使用上有不少改動(只是API函數名改了,而使用流程思維不變)，故在此記錄筆記存檔！

一.基本用法

1.創(chuàng)建Observable(被觀察者/發(fā)布者/發(fā)射者)

(1)create()
Observable observable = Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(@NonNull ObservableEmitter<String> observableEmitter) throws Exception {
        observableEmitter.onNext("發(fā)布數據1");
        observableEmitter.onNext("發(fā)布數據2");
        observableEmitter.onError(new Throwable("發(fā)生錯誤"));          
        observableEmitter.onComplete(); //完成
    }
});

(2)just
Observable observable = Observable.just("發(fā)布數據1", "發(fā)布數據2");

(3)fromIterable, fromArray
ArrayList<String> list = new ArrayList<>();
list.add("發(fā)布數據1");
list.add("發(fā)布數據2");
Observable observable = Observable.fromIterable(list);

(4)range,第一個參數為起始值,第二個為發(fā)送的個數,如果為0則不發(fā)送，負數則拋異常
Observable observable = Observable.range(10, 5)

(5)defer,延期,有觀察者訂閱時才創(chuàng)建Observable
Observable observable = Observable.defer(new Callable<ObservableSource>() {
    @Override
    public ObservableSource call() throws Exception {
        return Observable.just("發(fā)布數據1","發(fā)布數據2");
    }
});

(6)interval,定時周期發(fā)布數據
 Observable observable = Observable.interval(500, TimeUnit.MILLISECONDS); //周期500ms

(7)timer,延遲發(fā)布數據
Observable observable = Observable.timer(300, TimeUnit.MILLISECONDS); //延遲300ms

(8)repeat,重復發(fā)布數據
Observable observable = Observable.just("發(fā)布數據1").repeat(3); //重復發(fā)布3次

2.創(chuàng)建Observer(觀察者/訂閱者/接收者)

(1).Observer完整形式
Observer observer = new Observer() {
    @Override
    public void onSubscribe(@NonNull Disposable disposable) {
        //Disposable 相當于RxJava1.x中的Subscription,用于解除訂閱
    }

    @Override
    public void onNext(@NonNull Object o) {
        //接收數據
    }

    @Override
    public void onError(@NonNull Throwable throwable) {
        //接收錯誤
    }

    @Override
    public void onComplete() {
        //通知完成
    }
};

(2).Observer簡寫形式
Consumer onNext = new Consumer() {//接收數據
    @Override
    public void accept(Object o) throws Exception {
    }
};

Consumer<Throwable> onError = new Consumer<Throwable>() {//接收錯誤
    @Override
    public void accept(Throwable throwable) throws Exception {
    }
};

Action onComplete = new Action() {//通知完成
    @Override
    public void run() throws Exception {
    }
};

Consumer<Disposable> onSubscribe = new Consumer<Disposable>() {
    @Override
    public void accept(Disposable disposable) throws Exception {
    }
};

3.Observer訂閱Observable

(1).Observer完整訂閱
observable.subscribe(observer); //訂閱

(2).Observer簡寫訂閱
observable.subscribe(onNext);
observable.subscribe(onNext, onError);
observable.subscribe(onNext, onError, onComplete);
observable.subscribe(onNext, onError, onComplete, onSubscribe);

二.線程調度

調度器類型
Schedulers.computation(?)  用于計算任務，如事件循環(huán)或和回調處理，不要用于IO操作(IO操作請使用Schedulers.io())；默認線程數等于處理器的數量
Schedulers.from(executor)  使用指定的Executor作為調度器
Schedulers.io(?)           用于IO密集型任務，如異步阻塞IO操作，這個調度器的線程池會根據需要增長；
                           對于普通的計算任務，請使用Schedulers.computation()；
                           Schedulers.io(?)默認是一個CachedThreadScheduler，很像一個有線程緩存的新線程調度器
Schedulers.newThread(?)    為每個任務創(chuàng)建一個新線程
Schedulers.trampoline(?)   當其它排隊的任務完成后，在當前線程排隊開始執(zhí)行
AndroidSchedulers.mainThread()  此調度器為RxAndroid特有，顧名思義，運行在Android UI線程上

Observable.just("耗時操作...")
    .subscribeOn(Schedulers.io())//io線程-發(fā)布者
    .observeOn(AndroidSchedulers.mainThread())//主線程-接收者
    .subscribe(new Consumer<String>() {
        @Override
        public void accept(String s) throws Exception {                        
        }
    });

三.常用操作符

1.map-數據類型轉換

Observable.just("123")
    .map(new Function<String, Integer>() {
        @Override
        public Integer apply(@NonNull String s) throws Exception {
            return Integer.parseInt(s);
        }
    })
    .subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println(integer);
        }
    });

2.flatMap-數據集合扁平化(遍歷循環(huán)每一個元素)

List<String> list = new ArrayList<>();
list.add("1");
list.add("2");
list.add("3");
List<List<String>> listSSS = new ArrayList<>();//二維數組集合
listSSS.add(list);
Observable.fromIterable(listSSS)
.flatMap(new Function<List<String>, ObservableSource<String>>() {
    @Override
    public ObservableSource<String> apply(@NonNull List<String> list) throws Exception {
        return Observable.fromIterable(list);
    }
})
.subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        System.out.println(s);
    }
});

3.buffer-緩存滿后,以list集合發(fā)送數據

List<String> list = new ArrayList<>();
list.add("1");
list.add("2");
list.add("3");
Observable.fromIterable(list)
    .buffer(list.size())  //緩存一起發(fā)送
    .subscribe(new Consumer<List<String>>() {
        @Override
        public void accept(List<String> list) throws Exception {
            System.out.println(list.size());
        }
    });

4.take(n)-發(fā)送前n項數據

Observable.just(1, 2, 1, 1, 2, 3)
    .take(3) //發(fā)送前3項數據
    .subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println(integer);
        }
    });

5.distinct-去除重復項

Observable.just(1, 2, 1, 1, 2, 3)
    .distinct() //去重
    .subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println(integer);
        }
    });

6.filter-過濾

Observable.just(1, 2, 3, 4, 5)
    .filter(new Predicate<Integer>() {
        @Override
        public boolean test(@NonNull Integer integer) throws Exception {
            return integer > 3; //過濾大于3
        }
    })
    .subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println(integer);
        }
    });

五.Flowable-背壓

Flowable是RxJava2.x中新增的類，專門用于應對背壓Backpressure問題
背壓: 即生產者的速度大于消費者的速度帶來的問題，比如在Android中常見的點擊事件，點擊過快則會造成點擊兩次的效果!
Flowable.create(new FlowableOnSubscribe<Integer>() {
        @Override
        public void subscribe(FlowableEmitter<Integer> e) throws Exception {
            for (int i = 0; i < 10000; i++)
                e.onNext(i);
            e.onComplete();
        }
}, BackpressureStrategy.ERROR) //指定背壓處理策略，拋出異常錯
    .subscribeOn(Schedulers.computation())
    .observeOn(Schedulers.newThread())
    .subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println(integer);
            Thread.sleep(1000);
        }
    }, new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            System.out.println(throwable);
        }
    });

// 如Rxjava1.x一樣簡寫
Flowable.range(1,10000)
.onBackpressureDrop() // 背壓
.subscribe(new Consumer<Integer>() {
    @Override
    public void accept(Integer integer) throws Exception {
        System.out.println(integer);
    }
});

四.Subject

Subject extends Observable implements Observe
作用:
    可充當Observable
    可充當Observer
    是Observable和Observer之間的橋梁        
Subject有四個實現類: AsyncSubject, BehaviorSubject, PublishSubject, ReplaySubject
注意：
    從多個線程中調用onNext(on系列方法),需要使用串行化Serialized,才能順序調用！
    SerializedSubject<String, Integer> ser = new SerializedSubject(publishSubject);

Processor和Subject的作用相同,其中Processor是RxJava2.x新增的,繼承自Flowable,所以支持背壓控制
//Processor
AsyncProcessor<String> processor = AsyncProcessor.create();
processor.subscribe(o -> Log.d("JG",o)); //three
processor.onNext("one");
processor.onNext("two");
processor.onNext("three");
processor.onComplete();

1.AsyncSubject只接收onCompleted()被調用前的最后一個數據

AsyncSubject<String> asyncSubject = AsyncSubject.create();
asyncSubject.onNext("asyncSubject1");
asyncSubject.onNext("asyncSubject2");
asyncSubject.onNext("asyncSubject3");
asyncSubject.onComplete();
asyncSubject.subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        System.out.println(s);//只接收到asyncSubject3
    }
});

2.BehaviorSubject接收被訂閱前的最后一個數據,還接收訂閱后的數據

BehaviorSubject<String> behaviorSubject = BehaviorSubject.create();
behaviorSubject.onNext("behaviorSubject1");
behaviorSubject.onNext("behaviorSubject2");
behaviorSubject.subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        System.out.println(s); //接收到behaviorSubject2, behaviorSubject3, behaviorSubject4
    }
});
behaviorSubject.onNext("behaviorSubject3");
behaviorSubject.onNext("behaviorSubject4");

3.PublishSubject只接收被訂閱后的數據

PublishSubject<String> publishSubject = PublishSubject.create();
publishSubject.onNext("publishSubject1");
publishSubject.onNext("publishSubject2");
publishSubject.subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        System.out.println(s);只接收到behaviorSubject3, behaviorSubject4
    }
});
publishSubject.onNext("publishSubject3");
publishSubject.onNext("publishSubject4");

4.ReplaySubject接收所有數據,無論何時訂閱! 但緩存到一定大小時或一段時間后會丟棄舊的數據！

ReplaySubject<String> replaySubject = ReplaySubject.create(); //默認初始緩存容量大小為16
//replaySubject = ReplaySubject.create(100);//指定初始緩存容量大小為100
//replaySubject = ReplaySubject.createWithSize(2);//只緩存訂閱前最后2條數據
//replaySubject = ReplaySubject.createWithTime(1,TimeUnit.SECONDS,Schedulers.computation());//只緩存被訂閱前1秒內的數據
replaySubject.onNext("replaySubject:pre1");
replaySubject.onNext("replaySubject:pre2");
replaySubject.onNext("replaySubject:pre3");
replaySubject.subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        System.out.println(s);
    }
});
replaySubject.onNext("replaySubject:after1");
replaySubject.onNext("replaySubject:after2");

5.Subject作為橋梁,使用示例

//1.Subject作為橋梁
Subject<String> subject = BehaviorSubject.create();

//2.訂閱
subject.subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        System.out.println(s);
    }
});

//3.發(fā)布數據
Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(@NonNull ObservableEmitter<String> observableEmitter) throws Exception {
        observableEmitter.onNext("as Bridge");
    }
}).subscribe(subject);

簡書: http://www.itdecent.cn/p/724c937e3d0c
CSDN博客: http://blog.csdn.net/qq_32115439/article/details/78090944
GitHub博客: http://lioil.win/2017/09/25/JavaSE-RxJava.html
Coding博客: http://c.lioil.win/2017/09/25/JavaSE-RxJava.html

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av

Java-RxJava2筆記

Java-RxJava2筆記

一.基本用法

1.創(chuàng)建Observable(被觀察者/發(fā)布者/發(fā)射者)

2.創(chuàng)建Observer(觀察者/訂閱者/接收者)

3.Observer訂閱Observable

二.線程調度

三.常用操作符

1.map-數據類型轉換

2.flatMap-數據集合扁平化(遍歷循環(huán)每一個元素)

3.buffer-緩存滿后,以list集合發(fā)送數據

4.take(n)-發(fā)送前n項數據

5.distinct-去除重復項

6.filter-過濾

五.Flowable-背壓

四.Subject

1.AsyncSubject只接收onCompleted()被調用前的最后一個數據

2.BehaviorSubject接收被訂閱前的最后一個數據,還接收訂閱后的數據

3.PublishSubject只接收被訂閱后的數據

4.ReplaySubject接收所有數據,無論何時訂閱! 但緩存到一定大小時或一段時間后會丟棄舊的數據！

5.Subject作為橋梁,使用示例

相關閱讀更多精彩內容

友情鏈接更多精彩內容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九 欧美,1769亚洲,黄色成人av

Java-RxJava2筆記

一.基本用法

1.創(chuàng)建Observable(被觀察者/發(fā)布者/發(fā)射者)

2.創(chuàng)建Observer(觀察者/訂閱者/接收者)

3.Observer訂閱Observable

二.線程調度

三.常用操作符

1.map-數據類型轉換

2.flatMap-數據集合扁平化(遍歷循環(huán)每一個元素)

3.buffer-緩存滿后,以list集合發(fā)送數據

4.take(n)-發(fā)送前n項數據

5.distinct-去除重復項

6.filter-過濾

五.Flowable-背壓

四.Subject

1.AsyncSubject只接收onCompleted()被調用前的最后一個數據

2.BehaviorSubject接收被訂閱前的最后一個數據,還接收訂閱后的數據

3.PublishSubject只接收被訂閱后的數據

4.ReplaySubject接收所有數據,無論何時訂閱! 但緩存到一定大小時或一段時間后會丟棄舊的數據！

5.Subject作為橋梁,使用示例

相關閱讀更多精彩內容

友情鏈接更多精彩內容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av