RxJava2輕鬆入門

時間  2019-11-16
標籤 rxjava2 rxjava 輕鬆 入門 欄目 Java 简体版
原文   原文鏈接

Demo地址

RxJava2Demojavascript

本Demo旨在幫助從未接觸過RxJava的同窗直接入坑RxJava2，如絲般順滑，萬水千山老是情，留個star行不行？java

RxJava & RxAndroid （2.0版）

定義

RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.react

一個在 Java VM 上使用可觀測的序列來組成異步的、基於事件的程序的庫android

初學者若是看到這個準確但晦澀的定義確定一臉懵逼，不過咱們咱們只要把握重點便可：git

  • 異步
  • 基於事件
  • 觀察者模式

RxAndroid - Android specific bindings for RxJava 2.This module adds the minimum classes to RxJava that make writing reactive components in Android applications easy and hassle-free.github

RxAndroid在RxJava的基礎上添加了最少的類使得開發Android應用中的響應式組件更加的容易和自由api

特色

簡潔，並非指代碼量上的那種簡潔，而是邏輯上的簡潔，隨着程序邏輯變得愈來愈複雜，它依然可以保持簡潔。網絡

Github

Hello world

添加依賴

compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile 'io.reactivex.rxjava2:rxjava:2.0.1'複製代碼

簡單版本

//簡單版本
private void helloWorldSimple() {
    //建立消費者，消費者接受一個String類型的事件
    Consumer<String> consumer = new Consumer<String>() {
        @Override
        public void accept(String s) throws Exception {
            Log.d(TAG, s);
        }
    };
    //被觀察者發出Hello World, 而且指定該事件的消費者爲consumer
    Observable.just("Hello World").subscribe(consumer);
}複製代碼

運行結果

D/MainActivity: Hello World複製代碼

複雜版本

private void helloWorldComplex() {
    //Observer能夠看作Consumer的完整版
    Observer<String> observer = new Observer<String>() {

        //當Observable調用subscribe方法時會回調該方法
        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, "onSubscribe: ");
        }

        //onSubscribe方法後調用
        @Override
        public void onNext(String value) {
            Log.d(TAG, "onNext: " + value);
        }
        //這裏沒有出錯，沒有被調用
        @Override
        public void onError(Throwable e) {
            Log.d(TAG, "onError: ");
        }

        //onNext以後調用
        @Override
        public void onComplete() {
            Log.d(TAG, "onComplete: ");
        }
    };
    //被觀察者發出Hello World, 而且指定該事件的觀察者爲observer
    Observable.just("Hello World").subscribe(observer);
}複製代碼

運行結果

D/MainActivity: onSubscribe: 
D/MainActivity: onNext: Hello World
D/MainActivity: onComplete: 複製代碼

變態版本

private void helloWorldPlus() {
    //建立一個觀察者
    Observer<String> observer = new Observer<String>() {

        //當Observable調用subscribe方法時會回調該方法
        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, "onSubscribe: ");
        }

        //onSubscribe方法後調用
        @Override
        public void onNext(String value) {
            Log.d(TAG, "onNext: " + value);
        }
        //這裏沒有出錯，沒有被調用
        @Override
        public void onError(Throwable e) {
            Log.d(TAG, "onError: ");
        }

        //onNext以後調用
        @Override
        public void onComplete() {
            Log.d(TAG, "onComplete: ");
        }
    };

    //建立一個Observable
    Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {

        @Override
        public void subscribe(ObservableEmitter<String> e) throws Exception {
            e.onNext("Hello World");//會調用到觀察者的onNext
            e.onComplete();//會調用到觀察者的onComplete
        }
    });

    observable.subscribe(observer);
}複製代碼

運行結果

D/MainActivity: onSubscribe: 
D/MainActivity: onNext: Hello World
D/MainActivity: onComplete: 複製代碼

filter操做符

你早上去吃早餐，師傅是被觀察者，說咱這有包子，饅頭，腸粉，春捲，餃子，炒粉，你仔細想了想，發現你是最喜歡餃子的，因此把其餘的都排除掉，
因而你就吃到了餃子。app

private void filter() {
    //把Consumer能夠看作精簡版的Observer
    Consumer<String> consumer = new Consumer<String>() {
        //accept能夠簡單的看作onNext
        @Override
        public void accept(String s) throws Exception {
            Log.d(TAG, "accept: " + s);//這裏只能吃上餃子
        }
    };

    Observable.just("包子", "饅頭", "腸粉", "春捲", "餃子", "炒粉")
            .filter(new Predicate<String>() {
                @Override
                public boolean test(String s) throws Exception {
                    Log.d(TAG, "test: " + s);
                    return s.equals("餃子");//只容許餃子經過測試
                }
            })
            .subscribe(consumer);
}複製代碼

運行結果

D/MainActivity: test: 包子
D/MainActivity: test: 饅頭
D/MainActivity: test: 腸粉
D/MainActivity: test: 春捲
D/MainActivity: test: 餃子
D/MainActivity: accept: 餃子
D/MainActivity: test: 炒粉複製代碼

map操做符

map操做符可以完成數據類型的轉換。 如下代碼展現了一個Student到Developer的轉換。異步

private void map() {
    Observer<Developer> observer = new Observer<Developer>() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, "onSubscribe: ");
        }

        //觀察者接收到一個Developer
        @Override
        public void onNext(Developer value) {
            Log.d(TAG, "onNext: "  + value.toString());
        }

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

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

    Student student = new Student();
    student.setName("Leon");
    student.setAge(18);
    //map操做符，從Student類型轉換成Developer
    Observable.just(student).map(new Function<Student, Developer>() {
        @Override
        public Developer apply(Student student) throws Exception {
            Log.d(TAG, "apply: " + student.toString());
            Developer developer = new Developer();
            developer.setName(student.getName());
            developer.setAge(student.getAge());
            developer.setSkill("Android");
            return developer;
        }
    }).subscribe(observer);
}複製代碼

運行結果

D/MainActivity: onSubscribe: 
D/MainActivity: apply: Student{name='Leon', age=18}
D/MainActivity: onNext: Developer{name='Leon', age=18, skill='Android'}
D/MainActivity: onComplete: 複製代碼

flatmap操做符

flatmap可以鏈式地完成數據類型的轉換和加工。

遍歷一個學校中全部班級全部學生

private void flatmapClassToStudent() {

    Observable.fromIterable(new School().getClasses())

            //輸入是Class類型，輸出是ObservableSource<Student>類型
            .flatMap(new Function<Class, ObservableSource<Student>>() {

                //輸入是Class類型，輸出是ObservableSource<Student>類型
                @Override
                public ObservableSource<Student> apply(Class aClass) throws Exception {
                    Log.d(TAG, "apply: " + aClass.toString());
                    return Observable.fromIterable(aClass.getStudents());
                }
            }).subscribe(
            new Observer<Student>() {
                @Override
                public void onSubscribe(Disposable d) {
                    Log.d(TAG, "onSubscribe: ");
                }

                @Override
                public void onNext(Student value) {
                    Log.d(TAG, "onNext: " + value.toString());
                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onComplete() {

                }
            });
}複製代碼

運行結果

D/MainActivity: onSubscribe: 
D/MainActivity: apply: Class0
D/MainActivity: onNext: Student{name='Class0_0', age=18}
D/MainActivity: onNext: Student{name='Class0_1', age=18}
D/MainActivity: onNext: Student{name='Class0_2', age=18}
D/MainActivity: apply: Class1
D/MainActivity: onNext: Student{name='Class1_0', age=18}
D/MainActivity: onNext: Student{name='Class1_1', age=18}
D/MainActivity: onNext: Student{name='Class1_2', age=18}
D/MainActivity: apply: Class2
D/MainActivity: onNext: Student{name='Class2_0', age=18}
D/MainActivity: onNext: Student{name='Class2_1', age=18}
D/MainActivity: onNext: Student{name='Class2_2', age=18}複製代碼

遍歷一個學校全部班級全部組的全部學生

private void flatmapClassToGroupToStudent() {
    Observable.fromIterable(new School().getClasses())
            //輸入是Class類型，輸出是ObservableSource<Group>類型
            .flatMap(new Function<Class, ObservableSource<Group>>() {
                @Override
                public ObservableSource<Group> apply(Class aClass) throws Exception {
                    Log.d(TAG, "apply: " + aClass.toString());
                    return Observable.fromIterable(aClass.getGroups());
                }
            })
            //輸入類型是Group，輸出類型是ObservableSource<Student>類型
            .flatMap(new Function<Group, ObservableSource<Student>>() {
                @Override
                public ObservableSource<Student> apply(Group group) throws Exception {
                    Log.d(TAG, "apply: " + group.toString());
                    return Observable.fromIterable(group.getStudents());
                }
            })
            .subscribe(
                    new Observer<Student>() {
                        @Override
                        public void onSubscribe(Disposable d) {
                            Log.d(TAG, "onSubscribe: ");
                        }

                        @Override
                        public void onNext(Student value) {
                            Log.d(TAG, "onNext: " + value.toString());
                        }

                        @Override
                        public void onError(Throwable e) {

                        }

                        @Override
                        public void onComplete() {

                        }
                    });
}複製代碼

運行結果

D/MainActivity: onSubscribe: 
D/MainActivity: apply: Class0
D/MainActivity: apply: Group0
D/MainActivity: onNext: Student{name='Group0_0', age=18}
D/MainActivity: onNext: Student{name='Group0_1', age=18}
D/MainActivity: onNext: Student{name='Group0_2', age=18}
D/MainActivity: apply: Group1
D/MainActivity: onNext: Student{name='Group1_0', age=18}
D/MainActivity: onNext: Student{name='Group1_1', age=18}
D/MainActivity: onNext: Student{name='Group1_2', age=18}
D/MainActivity: apply: Group2
D/MainActivity: onNext: Student{name='Group2_0', age=18}
D/MainActivity: onNext: Student{name='Group2_1', age=18}
D/MainActivity: onNext: Student{name='Group2_2', age=18}
D/MainActivity: apply: Class1
D/MainActivity: apply: Group0
D/MainActivity: onNext: Student{name='Group0_0', age=18}
D/MainActivity: onNext: Student{name='Group0_1', age=18}
D/MainActivity: onNext: Student{name='Group0_2', age=18}
D/MainActivity: apply: Group1
D/MainActivity: onNext: Student{name='Group1_0', age=18}
D/MainActivity: onNext: Student{name='Group1_1', age=18}
D/MainActivity: onNext: Student{name='Group1_2', age=18}
D/MainActivity: apply: Group2
D/MainActivity: onNext: Student{name='Group2_0', age=18}
D/MainActivity: onNext: Student{name='Group2_1', age=18}
D/MainActivity: onNext: Student{name='Group2_2', age=18}
D/MainActivity: apply: Class2
D/MainActivity: apply: Group0
D/MainActivity: onNext: Student{name='Group0_0', age=18}
D/MainActivity: onNext: Student{name='Group0_1', age=18}
D/MainActivity: onNext: Student{name='Group0_2', age=18}
D/MainActivity: apply: Group1
D/MainActivity: onNext: Student{name='Group1_0', age=18}
D/MainActivity: onNext: Student{name='Group1_1', age=18}
D/MainActivity: onNext: Student{name='Group1_2', age=18}
D/MainActivity: apply: Group2
D/MainActivity: onNext: Student{name='Group2_0', age=18}
D/MainActivity: onNext: Student{name='Group2_1', age=18}
D/MainActivity: onNext: Student{name='Group2_2', age=18}複製代碼

線程調度

關於RxJava的線程調度，初學者只須要掌握兩個api就夠夠的啦。

subscribeOn

指定Observable在一個指定的線程調度器上建立。只能指定一次，若是指定屢次則以第一次爲準

observeOn

指定在事件傳遞，轉換，加工和最終被觀察者接受發生在哪個線程調度器。可指定屢次，每次指定完都在下一步生效。

經常使用線程調度器類型

  • Schedulers.single() 單線程調度器，線程可複用
  • Schedulers.newThread() 爲每一個任務建立新的線程
  • Schedulers.io() 處理io密集型任務，內部是線程池實現，可自動根據需求增加
  • Schedulers.computation() 處理計算任務，如事件循環和回調任務
  • AndroidSchedulers.mainThread() Android主線程調度器

示例

private void scheduleThreads() {
    Observable.create(
            new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(ObservableEmitter<String> e) throws Exception {
                    Log.d(TAG, "subscribe: " + Thread.currentThread().getName());
                    e.onNext("Hello Leon Fan");
                    e.onComplete();
                }
            })
            //指定subscribe方法在io線程池中調用
            .subscribeOn(Schedulers.io())
            //指定onNext方法 onComplete的方法在新建的線程中調用
            .observeOn(Schedulers.newThread())
            .subscribe(
                    new Observer<String>() {
                        @Override
                        public void onSubscribe(Disposable d) {
                            Log.d(TAG, "onSubscribe: " + Thread.currentThread().getName());
                        }

                        @Override
                        public void onNext(String value) {
                            Log.d(TAG, "onNext: " + Thread.currentThread().getName() + " " + value);
                        }

                        @Override
                        public void onError(Throwable e) {

                        }

                        @Override
                        public void onComplete() {
                            Log.d(TAG, "onComplete: " + Thread.currentThread().getName());
                        }
                    });

}複製代碼

運行結果

D/MainActivity: onSubscribe: main
D/MainActivity: subscribe: RxCachedThreadScheduler-4
D/MainActivity: onNext: RxNewThreadScheduler-1 Hello Leon Fan
D/MainActivity: onComplete: RxNewThreadScheduler-1複製代碼

若是將示例中的.observeOn(Schedulers.newThread())改爲AndroidSchedulers.mainThread()，則運行結果以下：

D/MainActivity: onSubscribe: main
D/MainActivity: subscribe: RxCachedThreadScheduler-5
D/MainActivity: onNext: main Hello Leon Fan
D/MainActivity: onComplete: main複製代碼

RxJava與Retrofit集成

咱們作一個Demo經過網絡請求獲取豆瓣電影Top10的列表來展現RxJava和Retrofit的集成的姿式。

Retrofit集成

添加依賴

compile 'com.squareup.retrofit2:retrofit:2.1.0'
compile 'com.squareup.retrofit2:converter-gson:2.1.0'
//compile 'com.squareup.retrofit2:adapter-rxjava:2.1.0' 官方adapter僅支持rxjava1.0
compile 'com.jakewharton.retrofit:retrofit2-rxjava2-adapter:1.0.0'複製代碼

建立網絡接口

public interface Api {
    @GET("top250")
    Observable<MovieBean> listTop250(@Query("start") int start, @Query("count") int count);
}複製代碼

實現Api

public class MovieRetrofit {

    private static MovieRetrofit sMovieRetrofit;
    private final Api mApi;

    public static MovieRetrofit getInstance() {
        if (sMovieRetrofit == null) {
            synchronized (MovieRetrofit.class) {
                if (sMovieRetrofit == null) {
                    sMovieRetrofit = new MovieRetrofit();
                }
            }
        }
        return sMovieRetrofit;
    }

    private MovieRetrofit() {
        Retrofit retrofit = new Retrofit.Builder().baseUrl("https://api.douban.com/v2/movie/")
                .addConverterFactory(GsonConverterFactory.create())
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .build();
        mApi = retrofit.create(Api.class);
    }

    public Api getApi() {
        return mApi;
    }
}複製代碼

發送網絡請求刷新列表

<!--添加網絡權限-->
<uses-permission android:name="android.permission.INTERNET"/>

Observable<MovieBean> movieBeanObservable = MovieRetrofit.getInstance().getApi().listTop250(0, 10);
movieBeanObservable.subscribeOn(Schedulers.io())//在io線程池中執行map
        //將網絡的結果轉換成咱們要的電影名的列表
        .map(new Function<MovieBean, List<String>>() {
            @Override
            public List<String> apply(MovieBean movieBean) throws Exception {
                List<String> array = new ArrayList<String>();
                for (int i = 0; i < movieBean.getSubjects().size(); i++) {
                    String title = movieBean.getSubjects().get(i).getTitle();
                    array.add(title);
                }
                return array;
            }
        })
        .observeOn(AndroidSchedulers.mainThread())//在主線程中執行onNext
        .subscribe(new Observer<List<String>>() {

        ......

            @Override
            public void onNext(List<String> value) {
                ArrayAdapter<String> arrayAdapter = new ArrayAdapter<String>(MovieListActivity.this, android.R.layout.simple_list_item_1, value);
                setListAdapter(arrayAdapter);
            }
            ......
        });複製代碼

參考

本人旨在幫助從未接觸過RxJava的童鞋直接入坑RxJava2.0，更多使用姿式請自行參考其餘資料學習。

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程