RxAndroid 實踐快速入門

一、RxJava概念

RxJava官方定義

一個在 Java VM 上使用可觀測的序列來組成異步的、基於事件的程序的庫。用一個詞概括：異步，也就是說RxJava也可以理解為一個處理異步的類庫。Android也提供了處理異步的工具AsyncTask、Handler，當我們處理較復雜的異步時，異步代碼難寫且難以讀懂。RxJava相對於AsyncTask和Handler的好處就是讓異步處理代碼顯得更簡潔易懂。

RxJava的異步實現，是通過一種擴展的觀察者模式來實現的。

RxJava 的觀察者模式

RxJava 有四個基本概念：Observable (可觀察者，即被觀察者)、 Observer (觀察者)、 subscribe (訂閱)、事件。Observable 和Observer 通過 subscribe() 方法實現訂閱關系，從而 Observable 可以在需要的時候發出事件來通知 Observer。與傳統觀察者模式不同， RxJava 的事件回調方法除了普通事件 onNext()之外，還定義了兩個特殊的事件：onCompleted() 和 onError()。

onCompleted(): 事件隊列完結。RxJava不僅把每個事件單獨處理，還會把它們看做一個隊列。RxJava 規定，當不會再有新的onNext() 發出時，需要觸發 onCompleted() 方法作為標志。

onError(): 事件隊列異常。在事件處理過程中出異常時，onError()會被觸發，同時隊列自動終止，不允許再有事件發出。

 

二、RxJava實現過程

1) 創建 Observer

Observer 即觀察者，它決定事件觸發的時候將有怎樣的行為。 RxJava 中的 Observer 接口的實現方式：

 

Observer observer = new Observer() {
        @Override
        public void onCompleted() {
            
        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onNext(String s) {

        }
    };

除了 Observer 接口之外，RxJava 還內置了一個實現了 Observer 的抽象類：Subscriber。 Subscriber 對 Observer 接口進行了一些擴展，但他們的基本使用方式是完全一樣的：

 

 

Subscriber subscriber = new Subscriber() {
        @Override
        public void onCompleted() {
            
        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onNext(String s) {

        }
    };

 

 

2) 創建 Observable

Observable 即被觀察者，它決定什麼時候觸發事件以及觸發怎樣的事件。 RxJava 使用create()方法來創建一個 Observable ，並為它定義事件觸發規則：

 

Observable observable = Observable.create(new Observable.OnSubscribe() {
        @Override
        public void call(Subscriber subscriber) {
            
        }
    });

Observable常見的創建方式還有有如下3種：

 

fromCallable():方法傳入的參數類型是泛型。

 

Observable observable = Observable.fromCallable(new Callable() {
        @Override
        public String call() throws Exception {
            return null;
        }
    });

 

just():方法傳入的參數支持List、數組、多參數。

 

 

Observable> observer = Observable.just(getArrays());

 

from():方法傳入的參數支持List、數組、多參數。

 

 

Observable observable = Observable.from(getArrays());

 

 

3) Subscribe (訂閱)

創建了Observable和Observer之後，再用subscribe()方法將它們聯結起來，整條鏈子就可以工作了。

 

observable.subscribe(observer);

三、RxJava線程控制--Scheduler

在RxJava 中，Scheduler --調度器，相當於線程控制器，RxJava 通過它來指定每一段代碼應該運行在什麼樣的線程。RxJava 已經內置了幾個Scheduler，它們已經適合大多數的使用場景：

Schedulers.immediate(): 直接在當前線程運行，相當於不指定線程。這是默認的Scheduler。Schedulers.newThread(): 總是啟用新線程，並在新線程執行操作。Schedulers.io(): I/O 操作（讀寫文件、讀寫數據庫、網絡信息交互等）所使用的Scheduler。Schedulers.computation(): 計算所使用的Scheduler。這個計算指的是 CPU 密集型計算，即不會被 I/O 等操作限制性能的操作，例如圖形的計算。Android 還有一個專用的AndroidSchedulers.mainThread()，它指定的操作將在 Android 主線程運行。

 

有了這幾個Scheduler，就可以使用subscribeOn()和observeOn()兩個方法來對線程進行控制了。 subscribeOn(): 指定subscribe()所發生的線程，即Observable.OnSubscribe被激活時所處的線程。或者叫做事件產生的線程。 observeOn(): 指定Subscriber所運行在的線程。或者叫做事件消費的線程。

 

mSubcription = observerable
                //指定subscribe發生在io線程
                .subscribeOn(Schedulers.io())
                //指定subscribe回調發生在UI線程
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(List list) {
                        displayRecyclerView(list);
                    }
                });

 

四、RxJava變換

RxJava的變換有map,flatMap,concatMap,switchMap等等，這裡介紹一下最常用的map方法。

先來看一個例子(示例代碼中根據拼音搜索城市的例子)：

 

mSubscription = mPublishSubject
                .debounce(400, TimeUnit.MILLISECONDS)
                .observeOn(Schedulers.io())
                .map(new Func1>() {
                    @Override
                    public List call(String s) {
                        return serverHelper.searchCity(s);
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(List strings) {
                        handleSearchResults(strings);
                    }
                });

從這個例子可以看到map方法根據傳入的數據會返回自己想要的數據。可以理解為根據輸入返回輸出，數據處理的過程在map裡面發生，數據返回的格式由自己定義。map返回的數據傳送到subscribe訂閱的observer中的onNext方法中，在onNext方法中可以做界面數據綁定等操作。map裡面封裝的new Func1函數是帶有返回值的函數。也可以用下面的方式處理map返回結果：

 

 

mSubscription = mPublishSubject
                .debounce(400, TimeUnit.MILLISECONDS)
                .observeOn(Schedulers.io())
                .map(new Func1>() {
                    @Override
                    public List call(String s) {
                        return serverHelper.searchCity(s);
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1>() {
                    @Override
                    public void call(List strings) {
                        
                    }
                })

subscribe中封裝的new Action1函數即是對new Func1函數返回值的響應，它是不帶返回值的。可以在call方法裡面做一些UI操作。

 

細心的你可能還會發型一個方法，debounce方法。在這個例子中，程序會監聽用戶輸入的內容，當用戶每次輸入完，程序會異步查找匹配城市信息，為了避免一些不必要的匹配，這裡設置debounce的時間為400毫秒，意思是每次用戶輸入完畢400毫秒後程序異步執行開始匹配城市信息。

好了，關於RxJava的介紹到此就結束啦。

 

五、RxAndroid示例講解

1、RxAndroid包引入

 

    compile 'io.reactivex:rxandroid:1.1.0'
    compile 'io.reactivex:rxjava:1.1.0'

 

2、示例源碼介紹

Example1Activity源碼，RxJava簡單示例：

 

/**
 * RxAndroid同步加載示例
 */
public class Example1Activity extends Activity {

    private RecyclerView recyclerview;
    private SimpleStringAdapter adapter;
    private ProgressBar progressbar;


    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        configureLayout();
        createObservable();
    }

    private void createObservable() {
        Observable> observer = Observable.just(getArrays());
        observer.subscribe(new Observer>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(List strings) {
                adapter.setStrings(strings);
            }
        });
    }

    private void configureLayout() {
        setContentView(R.layout.activity1_rxandroid);
        recyclerview = (RecyclerView) findViewById(R.id.recyleview);
        progressbar = (ProgressBar) findViewById(R.id.loader);
        recyclerview.setLayoutManager(new LinearLayoutManager(Example1Activity.this));
        adapter = new SimpleStringAdapter(Example1Activity.this);
        recyclerview.setAdapter(adapter);
        progressbar.setVisibility(View.GONE);
    }

    private static List getArrays(){
        List list = new ArrayList<>();
        list.add("aaaa");
        list.add("bbbb");
        list.add("cccc");
        list.add("dddd");
        return list;
    }
}

 

 

Example4Activity源碼，RxJava異步加載，實現網絡數據請求，顯示到RecyclerView：

 

/**
 * RxAndroid異步加載示例
 */
public class Example4Activity extends Activity {

    private RecyclerView recyclerview;
    private BeautyAdapter adapter;
    private ProgressBar progressbar;

    private PublishSubject mPublishSubject;
    private Subscription mSubscription;

    private Page page;


    private List beautyList = new ArrayList<>();
    private int pageIndex = 1;
    private int pageSize = 10;
    private boolean isLastPage = false;
    private int lastVisibleItem = 0;
    private LinearLayoutManager linearLayoutManager;


    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        configureLayout();
        createObserver();

        recyclerview.setOnScrollListener(new RecyclerView.OnScrollListener() {
            @Override
            public void onScrolled(RecyclerView recyclerView, int dx, int dy) {
                super.onScrolled(recyclerView, dx, dy);
                lastVisibleItem = linearLayoutManager.findLastVisibleItemPosition();
            }

            @Override
            public void onScrollStateChanged(RecyclerView recyclerView, int newState) {
                super.onScrollStateChanged(recyclerView, newState);
                if (newState == RecyclerView.SCROLL_STATE_IDLE
                        && lastVisibleItem + 1 == adapter.getItemCount()) {
                    //根據類目網絡請求數據
                    if(!isLastPage){
                        page.setPageIndex(pageIndex);
                        mPublishSubject.onNext(page);
                    }
                }
            }
        });
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        if (mSubscription != null && !mSubscription.isUnsubscribed()) {
            mSubscription.unsubscribe();
        }
    }

    private void createObserver() {
        page = new Page(pageIndex,pageSize);
        mPublishSubject = PublishSubject.create();
        mSubscription = mPublishSubject
                //指定subscribe發生在io線程
                .observeOn(Schedulers.io())
                .map(new Func1>() {
                    @Override
                    public List call(Page page) {
                        //網絡數據請求
                        List list = ServerHelper.getBeautyList(page.getPageIndex(),page.getPageSize());
                        if(list.size() == pageSize){
                            pageIndex ++;
                            isLastPage = false;
                        }else if(list.size() < pageSize){
                            isLastPage = true;
                        }
                        return list;
                    }
                })
                //指定subscribe回調發生在UI線程
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(List beautyList) {
                        displayRecyclerView(beautyList);
                    }
                });
        mPublishSubject.onNext(page);
    }

    private void displayRecyclerView(List list) {
        beautyList.addAll(list);
        progressbar.setVisibility(View.GONE);
        adapter.notifyDataSetChanged();
    }

    private void configureLayout() {
        setContentView(R.layout.activity1_rxandroid);
        progressbar = (ProgressBar) findViewById(R.id.loader);
        recyclerview = (RecyclerView) findViewById(R.id.recyleview);
        linearLayoutManager = new LinearLayoutManager(Example4Activity.this, LinearLayoutManager.VERTICAL,false);
        recyclerview.setLayoutManager(linearLayoutManager);
        adapter = new BeautyAdapter(beautyList,Example4Activity.this);
        recyclerview.setAdapter(adapter);

        recyclerview.addOnItemTouchListener(new OnRecyclerViewClickListener(recyclerview) {
            @Override
            protected void onItemClick(RecyclerView.ViewHolder viewHolder) {
                Toast.makeText(Example4Activity.this,viewHolder.getLayoutPosition()+"",Toast.LENGTH_SHORT).show();
            }
        });
    }
}

 

 

Example3Acticity源碼，介紹了RxJava map的使用，根據用戶輸入城市拼音返回城市列表：

 

public class Example3Activity extends Activity {

    private EditText mSearchInput;
    private TextView mNoResultsIndicator;
    private RecyclerView mSearchResults;
    private SimpleStringAdapter mSearchResultsAdapter;

    private PublishSubject mPublishSubject;
    private Subscription mSubscription;

    private ServerHelper serverHelper;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        configureLayout();
        createObservables();
        listenToSearchInput();
    }

    private void listenToSearchInput() {
        mSearchInput.addTextChangedListener(new TextWatcher() {
            @Override
            public void beforeTextChanged(CharSequence s, int start, int count, int after) {

            }

            @Override
            public void onTextChanged(CharSequence s, int start, int before, int count) {
                mPublishSubject.onNext(s.toString());
            }

            @Override
            public void afterTextChanged(Editable s) {

            }
        });
    }

    private void createObservables() {
        mPublishSubject = PublishSubject.create();
        mSubscription = mPublishSubject
                .debounce(400, TimeUnit.MILLISECONDS)
                .observeOn(Schedulers.io())
                .map(new Func1>() {
                    @Override
                    public List call(String s) {
                        return serverHelper.searchCity(s);
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(List strings) {
                        handleSearchResults(strings);
                    }
                });
    }

    private void handleSearchResults(List cities) {
        if (cities.isEmpty()) {
            showNoSearchResults();
        } else {
            showSearchResults(cities);
        }
    }

    private void showNoSearchResults() {
        mNoResultsIndicator.setVisibility(View.VISIBLE);
        mSearchResults.setVisibility(View.GONE);
    }

    private void showSearchResults(List cities) {
        mNoResultsIndicator.setVisibility(View.GONE);
        mSearchResults.setVisibility(View.VISIBLE);
        mSearchResultsAdapter.setStrings(cities);
    }


    private void configureLayout() {
        setContentView(R.layout.activity_3_example);
        mSearchInput = (EditText) findViewById(R.id.search_input);
        mNoResultsIndicator = (TextView) findViewById(R.id.no_results_indicator);
        mSearchResults = (RecyclerView) findViewById(R.id.search_results);
        mSearchResults.setLayoutManager(new LinearLayoutManager(this));
        mSearchResultsAdapter = new SimpleStringAdapter(this);
        mSearchResults.setAdapter(mSearchResultsAdapter);

        serverHelper = new ServerHelper(Example3Activity.this);
    }


    @Override
    protected void onDestroy() {
        super.onDestroy();
        if (mSubscription != null && !mSubscription.isUnsubscribed()) {
            mSubscription.unsubscribe();
        }
    }
}

 

對了，每次事件注銷不要忘了注銷subscription：

 

@Override
    protected void onDestroy() {
        super.onDestroy();
        if (mSubscription != null && !mSubscription.isUnsubscribed()) {
            mSubscription.unsubscribe();
        }
    }

使用RxAndroid的感受，網絡請求數據加載顯示速度會更快，數據異步加載的過程被流程化，代碼更簡潔，邏輯更清晰。