RxAndroid 实践快速入门

Posted 2022-12-06 阿蛮家

一、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