资讯专栏INFORMATION COLUMN

RxJava学习(四利用RxJava打造自己的RxBus)

番茄西红柿 / 3295人阅读

摘要:前面说过的功能很强大,不仅仅是实现链式的异步操作,它的功能很强大还可以通过实现的消息事件传递功能,我们来看看使用实现继承了类又实现了接口,可以同时担当订阅者和被订阅者的角色一个类产生多个对象,用一存储起来,以进行批量的取消订阅。

前面说过Rxjava的功能很强大,不仅仅是实现链式的异步操作,它的功能很强大还可以通过RxBus实现EventBus的消息/事件传递功

能,我们来看看

 

RxBus

  1 package com.example.liujian.rxjavademo;
  2  
  3 import java.util.HashMap;
  4  
  5 import rx.Observable;
  6 import rx.Subscription;
  7 import rx.android.schedulers.AndroidSchedulers;
  8 import rx.functions.Action0;
  9 import rx.functions.Action1;
 10 import rx.schedulers.Schedulers;
 11 import rx.subjects.PublishSubject;
 12 import rx.subjects.SerializedSubject;
 13 import rx.subscriptions.CompositeSubscription;
 14  
 15 /**
 16  * @project_Name: RxjavaDemo
 17  * @package: com.example.liujian.rxjavademo
 18  * @description: 使用Rxjava,RxAndroid实现RxBus
 19  * @author: liujian
 20  * @date: 2016/10/5 11:08
 21  * @version: V1.0
 22  */
 23  
 24 public class RxBus {
 25     private static volatile  RxBus mInstance;
 26     //Subject继承了Observable类又实现了Observer接口, Subject可以同时担当订阅者和被订阅者的角色
 27     private SerializedSubject mSubject;
 28     //一个类产生多个Subscription对象,用一CompositeSubscription 存储起来,以进行批量的取消订阅。避免内存泄漏
 29     private HashMap mSubscriptionHashMap;
 30     private RxBus(){
 31         //Subject是非线程安全的,SerializedSubject将PublishSubject 转换成一个线程安全的Subject对象
 32         mSubject=new SerializedSubject<>(PublishSubject.create());
 33     }
 34     public static RxBus getInstance(){
 35         if(mInstance==null){
 36             synchronized (RxBus.class){
 37                 if(mInstance==null){
 38                     mInstance=new RxBus();
 39                 }
 40             }
 41         }
 42         return mInstance;
 43     }
 44     /**
 45      * 发生消息
 46      */
 47     public void post(Object o){
 48         mSubject.onNext(o);
 49     }
 50  
 51     /**
 52      * 返回指定类型的Observable实例
 53      * @param type:要处理的消息的类型
 54      * @param 
 55      * @return
 56      */
 57     public Observable toObservable(final Class type){
 58         return mSubject.ofType(type);
 59     }
 60  
 61     /**
 62      * 是否已含有观察者订阅
 63      * @return
 64      */
 65     public boolean hasObservers(){
 66         return mSubject.hasObservers();
 67     }
 68  
 69     /**
 70      * 默认的订阅方法
 71      * @param 
 72      * @return
 73      */
 74     public Subscription doSubscribe(Class type, Action1 next){
 75         return toObservable(type)
 76                 .subscribeOn(Schedulers.io())
 77                 .observeOn(AndroidSchedulers.mainThread())
 78                 .subscribe(next);
 79     }
 80     /**
 81      * 默认的订阅方法
 82      * @param 
 83      * @return
 84      */
 85     public Subscription doSubscribe(Class type, Action1 next,Action1 error){
 86         return toObservable(type)
 87                 .subscribeOn(Schedulers.io())
 88                 .observeOn(AndroidSchedulers.mainThread())
 89                 .subscribe(next,error);
 90     }
 91     /**
 92      * 默认的订阅方法
 93      * @param 
 94      * @return
 95      */
 96     public Subscription doSubscribe(Class type, Action1 next, Action1 error, Action0 complete){
 97         return toObservable(type)
 98                 .subscribeOn(Schedulers.io())
 99                 .observeOn(AndroidSchedulers.mainThread())
100                 .subscribe(next,error,complete);
101     }
102  
103     /**
104      * 保存订阅后的subscription,方便一次性取消订阅
105      * @param o
106      * @param subscription
107      */
108     public void addSubscription(Object o,Subscription subscription){
109         if(mSubscriptionHashMap==null){
110             mSubscriptionHashMap=new HashMap<>();
111         }
112         String key=o.getClass().getSimpleName();
113         if(mSubscriptionHashMap.containsKey(key)){
114             mSubscriptionHashMap.get(key).add(subscription);
115         }else{
116             CompositeSubscription compositeSubscription=new CompositeSubscription();
117             compositeSubscription.add(subscription);
118             mSubscriptionHashMap.put(key,compositeSubscription);
119         }
120     }
121  
122     /**
123      * 取消订阅
124      * @param o
125      */
126     public void unSubscribe(Object o){
127         if(mSubscriptionHashMap==null){
128             return;
129         }
130         String key=o.getClass().getSimpleName();
131         if(!mSubscriptionHashMap.containsKey(key)){
132             return;
133         }
134         if(mSubscriptionHashMap.get(key)!=null){
135             mSubscriptionHashMap.get(key).unsubscribe();
136         }
137         mSubscriptionHashMap.remove(key);
138     }
139 }
View Code

不多解释了,注释都写的很清楚了,我们来看一下如何使用

 

发送消息:

RxBus.getInstance().post("这是发送的消息");

处理消息:

public void doSubscribe(){
        Subscription subscription = RxBus.getInstance().doSubscribe(String.class, new Action1() {
            @Override
            public void call(String s) {
                Toast.makeText(MainActivity.this,s, Toast.LENGTH_SHORT).show();
            }
        });
        RxBus.getInstance().addSubscription(this,subscription);
    }

或者

public void doSubscribe(){
        Subscription subscribe = RxBus.getInstance().toObservable(String.class)
                .filter(new Func1() {
                    @Override
                    public Boolean call(String s) {
                        return true;
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1() {
                    @Override
                    public void call(String s) {
                        Log.i(TAG, "call: " + s);
                    }
                });
        RxBus.getInstance().addSubscription(this,subscribe);
    }

取消订阅:

 @Override
protected void onDestroy() {
        //取消订阅,释放内存
        RxBus.getInstance().unSubscribe(this);
        super.onDestroy();
    }  

功能的完善

     当我们使用PublishSubject时,可能有些功能还不是很完善,比如我们只能先订阅事件,然后发送事件,如果反过来,先发送了事件再进

行订阅操作,比如两个Activity之间传递消息,怎么保证发送的事件不丢失呢?也就是EventBus的StickEvent功能,这个时候

PublishSubject就没有办法实现,我们可以替换为BehaviorSubject

 private RxBus() {
    mSubject = new SerializedSubject<>(BehaviorSubject.create());
   }

   但是BehaviorSubject只能缓存最近一个发送给它的事件,如果我们需要缓存多个事件可以改用ReplaySubject 

文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。

转载请注明本文地址:https://www.ucloud.cn/yun/3086.html

相关文章

  • Android框架 - 收藏集 - 掘金

    摘要:说好的分类集合来啦,采用标签云的方式来展示掘金阅读提示点击下文中返回到顶部有分类不合理的地方请提。反编译这个后发现其使用个优质的开源项目掘金是由的开源的一个库,用于构建可预期的和声明式的用户界面。 想不想通过一线互联网公司面试? - Android - 掘金国内一线互联网公司内部面试题库 以下面试题来自于百度、小米、乐视、美团、58、猎豹、360、新浪、搜狐内部题库 熟悉本文中列出的知...

    zengdongbao 评论0 收藏0
  • Android框架 - 收藏集 - 掘金

    摘要:说好的分类集合来啦,采用标签云的方式来展示掘金阅读提示点击下文中返回到顶部有分类不合理的地方请提。反编译这个后发现其使用个优质的开源项目掘金是由的开源的一个库,用于构建可预期的和声明式的用户界面。 想不想通过一线互联网公司面试? - Android - 掘金国内一线互联网公司内部面试题库 以下面试题来自于百度、小米、乐视、美团、58、猎豹、360、新浪、搜狐内部题库 熟悉本文中列出的知...

    awesome23 评论0 收藏0
  • Rx系列学习 - 收藏集 - 掘金

    摘要:最近开始转移到掘金。要解决的问题支持切换网络请求地址,例如实现掘金前言以前在项目中使用,来进行事件通知与订阅。因此精华翻译掘金本人学习完了后,将值得重点注意的变化进行了翻译和归纳,适合正在使用的同学快速了解如果错误欢迎批评指正。 收藏安卓开发中非常实用优秀的库! 有图有真相! - Android - 掘金本来是打算收藏工具类的,但转念一想,已经有这么多优秀的库了,就没必要再去重复造轮子...

    kidsamong 评论0 收藏0
  • RxBus实现及简单使用

    摘要:相信大多数开发者都使用过,作为事件总线通信库,如果你的项目已经加入和,不妨用代替,以减少库的依赖。只会把在订阅发生的时间点之后来自原始的数据发射给观察者。 原文链接:http://lingyunzhu.github.io RxJava目前已经很火了,如果你尚未了解请看这里。对于RxJava这里不多做介绍。RxBus并不是一个库,而是一种模式。相信大多数开发者都使用过EventBus,作...

    Hancock_Xu 评论0 收藏0
  • 有用框架级东西

    摘要:多个可以提高你安卓开发技能的开源学习的最佳方式就是阅读,对程序员来说也是如此。如何做一个崩溃率少于千分之三噶应用章节列表这个系列的有关架构的概念,适合有大概三年以上经验的同学理解。 Android 开源图表库 - SmallChart SmallChart 是一个开源的 Android 图表库,拥有 5 种不同的图表类型来展示数据。可以通过输入的数据以及可设置的颜色、画笔大小等属性方便...

    MadPecker 评论0 收藏0

发表评论

0条评论

最新活动
阅读需要支付1元查看
<