Rxbus讲解

作者: Simplelove_f033 | 来源:发表于2019-03-07 12:41 被阅读92次

RxBus简介

什么是RxBus？

RxBus 名字看起来像一个库，但它并不是一个库，而是一种模式，它的思想是使用RxJava来实现了EventBus ，而让你不再需要使用 Otto 或者 GreenRobot 的 EventBus。------ 给 Android 开发者的 RxJava 详解
RxBus2.x就是基于RxJava2.x封装实现的类。

为什么要使用RxBus？

如上所说，我们可以通过封装RxJava实现EventBus。
随着RxJava在Android项目中的普及, 我们完全可以使用RxBus代替EventBus，减少库的引入，增加系统的稳定性。
EventBus虽然使用方便，但是在事件的生命周期的处理上需要我们利用订阅者的生命周期去注册和取消注册，这个部分还是略有麻烦之处。
而我们可以结合使用RxLifecycle来配置，简化这一步骤。
结合RxJava的强大能力和RxAndroid的进程调度，RxBus有更多更强大的功能。

Demo&代码

GitHub:https://github.com/homeven/Rxbus
RxBus实现

Gradle

implementation 'io.reactivex.rxjava2:rxjava:2.1.17' //RxJava
implementation 'io.reactivex.rxjava2:rxandroid:2.0.2' //RxAndroid

1
2
封装RxJava实现RxBus

Subject有两种用途:

做为observable向其他的observable发送事件
做为observer接收其他的observable发送的事件
我们要实现：事件总线、事件发布者以及事件订阅者。首先Subject既可以作为被观察者发送事件，也可以作为观察者接收事件，而RxJava内部的响应式的支持实现了事件总线的功能。
可以使用PublishSubject.create().toSerialized();生成一个Subject对象。
如下代码就实现了最基本的RxBus。
public class RxBus {
private volatile static RxBus mDefaultInstance;
private final Subject<Object> mBus;

        private RxBus() {
            mBus = PublishSubject.create().toSerialized();
        }

        public static RxBus getInstance() {
            if (mDefaultInstance == null) {
                synchronized (RxBus.class) {
                    if (mDefaultInstance == null) {
                        mDefaultInstance = new RxBus();
                    }
                }
            }
            return mDefaultInstance;
        }

        /**
         * 发送事件
         */
        public void post(Object event) {
            mBus.onNext(event);
        }

        /**
         * 根据传递的 eventType 类型返回特定类型(eventType)的 被观察者
         */
        public <T> Observable<T> toObservable(final Class<T> eventType) {
            return mBus.ofType(eventType);
        }

        /**
         * 判断是否有订阅者
         */
        public boolean hasObservers() {
            return mBus.hasObservers();
        }

        public void reset() {
            mDefaultInstance = null;
        }

    }

使用RxBus

事件实体

public class MsgEvent {
    private String msg;
    public MsgEvent(String msg) {
        this.msg = msg;
    }
    public String getMsg() {
        return msg;
    }
    public void setMsg(String msg) {
        this.msg = msg;
    }
}

订阅事件

    RxBus.getInstance().toObservable(MsgEvent.class).subscribe(new Observer<MsgEvent>() {
                @Override
                public void onSubscribe(Disposable d) {
                    
                }

                @Override
                public void onNext(MsgEvent msgEvent) {
                    //处理事件
                }

                @Override
                public void onError(Throwable e) {
                      
                }

                @Override
                public void onComplete() {

                }
            });

发送事件
RxBus.getInstance().post(new MsgEvent("Java"));

与EventBus对比，除了订阅事件的消息处理，使用基本一样。
至于EventBus的线程模型，我们完全可以使用RxJava 的线程控制Scheduler来实现，具体可以参考上面RxJava的使用两篇文章。

RxBus内存泄漏

使用RxJava发布一个订阅后，当页面被finish，此时订阅逻辑还未完成，如果没有及时取消订阅，就会导致Activity/Fragment无法被回收，从而引发内存泄漏。
EventBus为了解决这个问题，要求我们根据订阅者的生命周期注册和取消注册。所以在RxBus中我们也可以这样操作。

简单处理

使用一个CompositeDisposable存储当前所有的订阅，然后再onDestroy()中将其dispose()。

    private CompositeDisposable compositeDisposable;
     @Override
     protected void onCreate(Bundle savedInstanceState) {
    compositeDisposable = new CompositeDisposable();
            RxBus.getInstance().toObservable(MsgEvent.class).subscribe(new Observer<MsgEvent>() {
                @Override
                public void onSubscribe(Disposable d) {
                    compositeDisposable.add(d);
                }

                @Override
                public void onNext(MsgEvent msgEvent) {
                    //事件处理
                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onComplete() {

                }
            });
        }

        @Override
        public void onDestroy() {
            super.onDestroy();
            unbinder.unbind();
            if (compositeDisposable!=null && !compositeDisposable.isDisposed()){
                compositeDisposable.dispose();
            }
        }

RxLifecycle

Rxlifecycle 是trello开发的用于解决RxJava引起的内存泄漏的开源框架。
GitHub地址：https://github.com/trello/RxLifecycle
该框架为了适应不同的场景,开发了不同的版本，具体的可以查看GitHub文档:
// RxLifecycle基础库
implementation 'com.trello.rxlifecycle2:rxlifecycle:2.2.2'

// Android使用的库，可以绑定特定的生命周期
//需继承RxActvivty使用
implementation 'com.trello.rxlifecycle2:rxlifecycle-android:2.2.2'

// Android组件库，里面定义了例如RxAppCompatActivity、      RxFragment之类的Android组件
// 须继承RxAppCompatActivity、RxFragment使用
implementation 'com.trello.rxlifecycle2:rxlifecycle-components:2.2.2'

// 预先编写的支持首选项片段，将其子类化为提供者
implementation 'com.trello.rxlifecycle2:rxlifecycle-components-    preference:2.2.2'

// Android使用的库，须继承NaviActivity使用

implementation 'com.trello.rxlifecycle2:rxlifecycle-navi:2.2.2'

// 使用Android生命周期作为提供者
//无需继承，任何有声明周期的组件都可以直接使用
implementation 'com.trello.rxlifecycle2:rxlifecycle-android-    lifecycle:2.2.2'

// Kotlin语法 
implementation 'com.trello.rxlifecycle2:rxlifecycle-kotlin:2.2.2'

//在Kotlin语法，使用Android生命周期 
implementation 'com.trello.rxlifecycle2:rxlifecycle-android-    lifecycle-kotlin:2.2.2'

如果我们自己有BaseActivity，所以不能继承RxActvivty，RxAppCompatActivity、RxFragment，NaviActivity。
为了保持代码的灵活性，我们使用:

// 使用Android生命周期作为提供者
implementation 'com.trello.rxlifecycle2:rxlifecycle-android-lifecycle:2.2.2'

修改RxBus，添加如下代码：
使用compose(this.bindToLifecycle())方法绑定Activity的生命周期，在onStart方法中绑定，在onStop方法被调用后就会解除绑定，以此类推。
如果在onPause/onStop方法中绑定，那么就会在它的下一个生命周期方法（onStop/onDestory）被调用后解除绑定。

绑定销毁
onCreate onDestory
onStart onStop
onResum onPause
onPause onStop
onStop onDestory

 /**
     * 使用Rxlifecycle解决RxJava引起的内存泄漏
     */
    public <T> Observable<T> toObservable(LifecycleOwner owner, final Class<T> eventType) {
        LifecycleProvider<Lifecycle.Event> provider =     AndroidLifecycle.createLifecycleProvider(owner);
        return mBus.ofType(eventType).compose(provider.    <T>bindToLifecycle());
    }

订阅事件时，添加上下文this即可绑定生命周期，自动取消订阅:

        RxBus.getInstance().toObservable(this,MsgEvent.class).subscrib    e(new Consumer<MsgEvent>() {
            @Override
            public void accept(MsgEvent msgEvent) throws Exception {
                //处理事件
            }
        });

RxBus的粘性事件

EventBus有粘性事件，RxBus也可以实现。
修改RxBus修改&添加如下代码：

    private final Map<Class<?>, Object> mStickyEventMap;

        private RxBus() {
            mBus = PublishSubject.create().toSerialized();
            mStickyEventMap = new ConcurrentHashMap<>();
        }
      /**
         * 发送一个新Sticky事件
         */
        public void postSticky(Object event) {
            synchronized (mStickyEventMap) {
                mStickyEventMap.put(event.getClass(), event);
            }
            post(event);
        }

        /**
         * 根据传递的 eventType 类型返回特定类型(eventType)的 被观察者
         * 使用Rxlifecycle解决RxJava引起的内存泄漏
         */
        public <T> Observable<T> toObservableSticky(LifecycleOwner owner,final Class<T> eventType) {
            synchronized (mStickyEventMap) {
                LifecycleProvider<Lifecycle.Event> provider = AndroidLifecycle.createLifecycleProvider(owner);
                Observable<T> observable = mBus.ofType(eventType).compose(provider.<T>bindToLifecycle());
                final Object event = mStickyEventMap.get(eventType);

                if (event != null) {
                    return observable.mergeWith(Observable.create(new ObservableOnSubscribe<T>() {
                        @Override
                        public void subscribe(ObservableEmitter<T> subscriber) throws Exception {
                            subscriber.onNext(eventType.cast(event));
                        }
                    }));
                } else {
                    return observable;
                }
            }
        }

        /**
         * 根据eventType获取Sticky事件
         */
        public <T> T getStickyEvent(Class<T> eventType) {
            synchronized (mStickyEventMap) {
                return eventType.cast(mStickyEventMap.get(eventType));
            }
        }

        /**
         * 移除指定eventType的Sticky事件
         */
        public <T> T removeStickyEvent(Class<T> eventType) {
            synchronized (mStickyEventMap) {
                return eventType.cast(mStickyEventMap.remove(eventType));
            }
        }

        /**
         * 移除所有的Sticky事件
         */
        public void removeAllStickyEvents() {
            synchronized (mStickyEventMap) {
                mStickyEventMap.clear();
            }
        }

RxBus完整代码

    public class RxBus {
        private volatile static RxBus mDefaultInstance;
        private final Subject<Object> mBus;

        private final Map<Class<?>, Object> mStickyEventMap;

        private RxBus() {
            mBus = PublishSubject.create().toSerialized();
            mStickyEventMap = new ConcurrentHashMap<>();
        }

        public static RxBus getInstance() {
            if (mDefaultInstance == null) {
                synchronized (RxBus.class) {
                    if (mDefaultInstance == null) {
                        mDefaultInstance = new RxBus();
                    }
                }
            }
            return mDefaultInstance;
        }

        /**
         * 发送事件
         */
        public void post(Object event) {
            mBus.onNext(event);
        }

        /**
         * 使用Rxlifecycle解决RxJava引起的内存泄漏
         */
        public <T> Observable<T> toObservable(LifecycleOwner owner, final Class<T> eventType) {
            LifecycleProvider<Lifecycle.Event> provider = AndroidLifecycle.createLifecycleProvider(owner);
            return mBus.ofType(eventType).compose(provider.<T>bindToLifecycle());
        }

        /**
         * 判断是否有订阅者
         */
        public boolean hasObservers() {
            return mBus.hasObservers();
        }

        public void reset() {
            mDefaultInstance = null;
        }


        /**
         * Stciky 相关
         */

        /**
         * 发送一个新Sticky事件
         */
        public void postSticky(Object event) {
            synchronized (mStickyEventMap) {
                mStickyEventMap.put(event.getClass(), event);
            }
            post(event);
        }

        /**
         * 根据传递的 eventType 类型返回特定类型(eventType)的 被观察者
         * 使用Rxlifecycle解决RxJava引起的内存泄漏
         */
        public <T> Observable<T> toObservableSticky(LifecycleOwner owner,final Class<T> eventType) {
            synchronized (mStickyEventMap) {
                LifecycleProvider<Lifecycle.Event> provider = AndroidLifecycle.createLifecycleProvider(owner);
                Observable<T> observable = mBus.ofType(eventType).compose(provider.<T>bindToLifecycle());
                final Object event = mStickyEventMap.get(eventType);

                if (event != null) {
                    return observable.mergeWith(Observable.create(new ObservableOnSubscribe<T>() {
                        @Override
                        public void subscribe(ObservableEmitter<T> subscriber) throws Exception {
                            subscriber.onNext(eventType.cast(event));
                        }
                    }));
                } else {
                    return observable;
                }
            }
        }

        /**
         * 根据eventType获取Sticky事件
         */
        public <T> T getStickyEvent(Class<T> eventType) {
            synchronized (mStickyEventMap) {
                return eventType.cast(mStickyEventMap.get(eventType));
            }
        }

        /**
         * 移除指定eventType的Sticky事件
         */
        public <T> T removeStickyEvent(Class<T> eventType) {
            synchronized (mStickyEventMap) {
                return eventType.cast(mStickyEventMap.remove(eventType));
            }
        }

        /**
         * 移除所有的Sticky事件
         */
        public void removeAllStickyEvents() {
            synchronized (mStickyEventMap) {
                mStickyEventMap.clear();
            }
        }

    }

Demo

该Demo演示了使用RxBus完成Activty与Fragment的通信。

这里写图片描述

    public class RxActivity extends AppCompatActivity {

        @BindView(R.id.java)
        Button java;
        @BindView(R.id.android)
        Button android;
        @BindView(R.id.rx_layout)
        FrameLayout rxLayout;
        private FragmentManager fragmentManager;
        @Override
        protected void onCreate(Bundle savedInstanceState) {
            super.onCreate(savedInstanceState);
            setContentView(R.layout.activity_rx);
            ButterKnife.bind(this);
            fragmentManager = getSupportFragmentManager();
            FragmentTransaction transaction = fragmentManager.beginTransaction();
            transaction.add(R.id.rx_layout, new RxFragment());
            transaction.commit();
        }

        @OnClick({R.id.java, R.id.android})
        public void onViewClicked(View view) {
            switch (view.getId()) {
                case R.id.java:
                    RxBus.getInstance().post(new MsgEvent("Java"));
                    break;
                case R.id.android:
                    RxBus.getInstance().post(new MsgEvent("Android"));
                    break;
            }
        }
    }

    <?xml version="1.0" encoding="utf-8"?>
    <LinearLayout xmlns:android="http://schemas.android.com/apk/res/android"
        xmlns:app="http://schemas.android.com/apk/res-auto"
        xmlns:tools="http://schemas.android.com/tools"
        android:layout_width="match_parent"
        android:layout_height="match_parent"
        android:orientation="vertical">


        <LinearLayout
            android:layout_width="match_parent"
            android:layout_height="wrap_content">

            <Button
                android:id="@+id/java"
                android:layout_width="wrap_content"
                android:layout_height="wrap_content"
                android:text="Java" />

            <Button
                android:id="@+id/android"
                android:layout_width="wrap_content"
                android:layout_height="wrap_content"
                android:text="Android" />
        </LinearLayout>

        <FrameLayout
            android:id="@+id/rx_layout"
            android:layout_width="match_parent"
            android:layout_height="match_parent" />
    </LinearLayout>

    public class RxFragment extends Fragment {
        @BindView(R.id.text)
        TextView text;
        Unbinder unbinder;
        private View view;
        @SuppressLint("CheckResult")
        @Nullable
        @Override
        public View onCreateView(@NonNull LayoutInflater inflater, @Nullable ViewGroup container, @Nullable Bundle savedInstanceState) {
            view = inflater.inflate(R.layout.fragment_rx, container, false);
            unbinder = ButterKnife.bind(this, view);
        
            RxBus.getInstance().toObservable(this,MsgEvent.class).subscribe(new Consumer<MsgEvent>() {
                @Override
                public void accept(MsgEvent msgEvent) throws Exception {
                    //处理事件
                }
            });
            return view;
        }

    }

    <?xml version="1.0" encoding="utf-8"?>
    <LinearLayout xmlns:android="http://schemas.android.com/apk/res/android"
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    android:gravity="center_horizontal"
    android:orientation="vertical">

    <TextView
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:layout_marginTop="20dp"
        android:text="RxFragment"
        android:textSize="18sp" />

    <TextView
        android:id="@+id/text"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:layout_marginTop="20dp"
        android:textSize="16sp" />
    </LinearLayout>

参考

RxBus 简单使用学习
 https://www.2cto.com/kf/201708/674041.html

Rxbus讲解

使用RxBus

事件实体

订阅事件

Demo

参考

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

框架