RxAndroid 实践快速入门

时间：2016-05-27 11:51:13 阅读：500 评论：0 收藏：0 [点我收藏+]

标签：

一、RxJava概念

RxJava官方定义

一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库。用一个词概括：异步，也就是说RxJava也可以理解为一个处理异步的类库。Android也提供了处理异步的工具AsyncTask、Handler，当我们处理较复杂的异步时，异步代码难写且难以读懂。RxJava相对于AsyncTask和Handler的好处就是让异步处理代码显得更简洁易懂。

RxJava的异步实现，是通过一种扩展的观察者模式来实现的。什么是观察者模式可以参考Java设计模式之观察者模式。

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<String>() {
        @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<String>() {
        @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<String>() {
        @Override
        public void call(Subscriber<? super String> subscriber) {
            
        }
    });

Observable常见的创建方式还有有如下3种：

fromCallable():方法传入的参数类型是泛型。

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

just():方法传入的参数支持List、数组、多参数。

Observable<List<String>> 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<List<Beauty>>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

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

四、RxJava变换

RxJava的变换有map,flatMap,concatMap,switchMap等等，这里介绍一下最常用的map方法。

先来看一个例子(示例代码中根据拼音搜索城市的例子)：

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

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(List<String> 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<String, List<String>>() {
                    @Override
                    public List<String> call(String s) {
                        return serverHelper.searchCity(s);
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<List<String>>() {
                    @Override
                    public void call(List<String> 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<List<String>> observer = Observable.just(getArrays());
        observer.subscribe(new Observer<List<String>>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(List<String> 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<String> getArrays(){
        List<String> 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<Page> mPublishSubject;
    private Subscription mSubscription;

    private Page page;


    private List<Beauty> 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<Page, List<Beauty>>() {
                    @Override
                    public List<Beauty> call(Page page) {
                        //网络数据请求
                        List<Beauty> 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<List<Beauty>>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

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

    private void displayRecyclerView(List<Beauty> 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<String> 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<String, List<String>>() {
                    @Override
                    public List<String> call(String s) {
                        return serverHelper.searchCity(s);
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<List<String>>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

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

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

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

    private void showSearchResults(List<String> 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的感受，网络请求数据加载显示速度会更快，数据异步加载的过程被流程化，代码更简洁，逻辑更清晰。

文章参考自给 Android 开发者的 RxJava 详解关于RxJava作者写得很详细，有兴趣可以去看看，收获颇多哦。

项目参考 https://github.com/klnusbaum/rxandroidexamples

文章示例源码下载地址：https://github.com/xiaomanzijia/RxAndroidSample

RxAndroid 实践快速入门

标签：

原文地址：http://blog.csdn.net/u011459799/article/details/51500065

踩

(0)

评论一句话评论（0）

分享档案

更多>

2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)

周排行