RxBus 代替 EventBus 进行组件之间通讯

Bertha 。 2023-07-18 14:10 14阅读 0赞

**▲ 前言**：

事件总线的好处在于方便组件之间的交互，RxBus不是一个库，而是使用RxJava实现事件总线的一种思想。刚好项目中使用到了rxjava，相对于使用eventbus，使用rxbus来的更方便。

**▲ rxbus和eventbus相比较**：

*  RxJava 主要做异步、网络的数据处理，强大之处就是对数据的处理了，而对于处理完后的数据处理是一样的都是观察者模式来通知，也可以把 RxJava 进一步封装出一个 EventBus（RxBus） 库，二者可以转换的。
 *  EventBus比较适合仅仅当做组件间的通讯工具使用，主要用来传递消息。使用EventBus可以避免搞出一大推的interface，仅仅是为了实现组件间的通讯，而不得不去实现那一推的接口

**▲ rxbus实现原理**：

Note that it is important to subscribe to the exact same rxBus instance that was used to post the events

**▲ 基本配置**：

//引入rxJava
        compile 'io.reactivex.rxjava2:rxjava:2.1.8'
        //引入rxAndroid
        compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
        //引入rxJava适配器，方便rxJava与retrofit的结合
        compile 'com.jakewharton.retrofit:retrofit2-rxjava2-adapter:1.0.0'
        //引入J神的rxrelay2，出现异常仍然可以处理
        compile 'com.jakewharton.rxrelay2:rxrelay:2.0.0'
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8

**▲ 基本使用 一**：

在订阅者处理事件出现异常后，订阅者无法再收到事件，这是 RxJava 当初本身的设计原则，但是在事件总线中这反而是个问题，不过 JakeWharton 大神写了即使出现异常也不会终止订阅关系的 RxRelay，所以基于 RxRelay 就能写出有异常处理能力的 Rxbus。

/** * 有异常处理的 Rxbus * @author Donkor */
    
    public class RxBus { 
        private static volatile RxBus instance;
        private final Relay<Object> mBus;
    
        public RxBus() {
            this.mBus = PublishRelay.create().toSerialized();
        }
    
        public static RxBus getInstance() {
            if (instance == null) {
                synchronized (RxBus.class) {
                    if (instance == null) {
                        instance = Holder.BUS;
                    }
                }
            }
            return instance;
        }
        public void post(Object obj) {
            mBus.accept(obj);
        }
    
        public <T> Observable<T> toObservable(Class<T> tClass) {
            return  mBus.ofType(tClass);
        }
    
        public Observable<Object> toObservable() {
            return mBus;
        }
    
        public boolean hasObservers() {
            return mBus.hasObservers();
        }
    
        private static class Holder { 
            private static final RxBus BUS = new RxBus();
        }
    }
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36
       37
       38
       39
       40
       41
       42
       43

关于rxbus异常处理还不太了解的同学可以参考一下这篇文章：[\[深入RxBus\]：异常处理][RxBus]

**▲ 基本使用 二**：

没有背压处理（Backpressure）的 Rxbus

/** * @author Donkor */
    public class RxBus { 
    
        private final Subject<Object> mBus;
    
        private RxBus() {
            // toSerialized method made bus thread safe
            mBus = PublishSubject.create().toSerialized();
        }
    
        public static RxBus get() {
            return Holder.BUS;
        }
    
        public void post(Object obj) {
            mBus.onNext(obj);
        }
    
        public <T> Observable<T> toObservable(Class<T> tClass) {
            return mBus.ofType(tClass);
        }
    
        public Observable<Object> toObservable() {
            return mBus;
        }
    
        public boolean hasObservers() {
            return mBus.hasObservers();
        }
    
        private static class Holder { 
            private static final RxBus BUS = new RxBus();
        }
    }
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36

有背压处理（Backpressure）的 Rxbus

/** * @author Donkor */
    public class RxBus { 
    
        private final FlowableProcessor<Object> mBus;
    
        private RxBus() {
            // toSerialized method made bus thread safe
            mBus = PublishProcessor.create().toSerialized();
        }
    
        public static RxBus get() {
            return Holder.BUS;
        }
    
        public void post(Object obj) {
            mBus.onNext(obj);
        }
    
        public <T> Flowable<T> toFlowable(Class<T> tClass) {
            return mBus.ofType(tClass);
        }
    
        public Flowable<Object> toFlowable() {
            return mBus;
        }
    
        public boolean hasSubscribers() {
            return mBus.hasSubscribers();
        }
    
        private static class Holder { 
            private static final RxBus BUS = new RxBus();
        }
    }
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36

关于有背压与没有背压的知识介绍，可以参考下这一篇文章[关于RxJava背压][RxJava]

**▲ 举个栗子**：

场景：当前有三个（多个）activity，打开第二、三个（多个）activity需要修改第一个activity的数据时，这时就可以使用rxbus，相比发送广播速度快。

无图无真相，直接看下效果图  
![70][]

第一个页面关键代码

RxBus.getInstance().toObservable().map(new Function<Object, EventMsg>() {
                @Override
                public EventMsg apply(Object o) throws Exception {
                    return (EventMsg) o;
                }
            }).subscribe(new Consumer<EventMsg>() {
                @Override
                public void accept(EventMsg eventMsg) throws Exception {
                    if (eventMsg != null) {
                        mTvContent.setText(eventMsg.getMsg());
                    }
                }
            });
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13

第二个页面关键代码

EventMsg eventMsg = new EventMsg();
                    eventMsg.setMsg("来自第二个页面发送过来的数据 --- 修改成功");
                    RxBus.getInstance().post(eventMsg);
     
      
     
      
       1
       2
       3

EventMsg数据类

public class EventMsg {
        private String msg;
    
        public String getMsg() {
            return msg;
        }
    
        public void setMsg(String msg) {
            this.msg = msg;
        }
    }
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11

**▲ Demo下载地址**：**[https://download.csdn.net/download/donkor\_/10312846][https_download.csdn.net_download_donkor_10312846]**

[RxBus]: https://www.jianshu.com/p/0493cc28a811
[RxJava]: http://flyou.ren/2017/04/05/%E5%85%B3%E4%BA%8ERxJava%E8%83%8C%E5%8E%8B/?utm_source=tuicool&utm_medium=referral
[70]: /images/20230528/ba8f0fe171ed48109a8b2c2f78ebbac7.png
[https_download.csdn.net_download_donkor_10312846]: https://download.csdn.net/download/donkor_/10312846