Android项目中使用RxBus进行通讯

加入JakeWharton大神的RxRelay

JakeWharton/RxRelay

compile 'com.jakewharton.rxrelay2:rxrelay:2.0.0'

RxBus.java

/**
 * If you have any questions, you can contact by email {wangzhumoo@gmail.com}
 * @author 王诛魔 2017/11/30 下午6:09
 *
 */
public class RxBus {

    private Relay<Object> mBus = null;
    private static volatile RxBus mDefaultInstance;


    public RxBus() {
        mBus = PublishRelay.create().toSerialized();
    }

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

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

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


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


    /**
     * 取消订阅
     * @param disposable
     */
    public void unregister(Disposable disposable) {
        if (disposable != null && !disposable.isDisposed()) {
            disposable.dispose();
        }
    }


    public <T> Disposable register(Class<T> eventType, Scheduler scheduler, Consumer<T> onNext) {
        return toObservable(eventType).observeOn(scheduler).subscribe(onNext);
    }

    public <T> Disposable register(Class<T> eventType, Scheduler scheduler, Consumer<T> onNext, Consumer onError,
                                   Action onComplete, Consumer onSubscribe) {
        return toObservable(eventType).observeOn(scheduler).subscribe(onNext, onError, onComplete, onSubscribe);
    }

    public <T> Disposable register(Class<T> eventType, Scheduler scheduler, Consumer<T> onNext, Consumer onError,
                                   Action onComplete) {
        return toObservable(eventType).observeOn(scheduler).subscribe(onNext, onError, onComplete);
    }

    public <T> Disposable register(Class<T> eventType, Scheduler scheduler, Consumer<T> onNext, Consumer onError) {
        return toObservable(eventType).observeOn(scheduler).subscribe(onNext, onError);
    }

    public <T> Disposable register(Class<T> eventType, Consumer<T> onNext) {
        return toObservable(eventType).observeOn(AndroidSchedulers.mainThread()).subscribe(onNext);
    }

    public <T> Disposable register(Class<T> eventType, Consumer<T> onNext, Consumer onError, Action onComplete, Consumer onSubscribe) {
        return toObservable(eventType).observeOn(AndroidSchedulers.mainThread()).subscribe(onNext, onError, onComplete, onSubscribe);
    }

    public <T> Disposable register(Class<T> eventType, Consumer<T> onNext, Consumer onError, Action onComplete) {
        return toObservable(eventType).observeOn(AndroidSchedulers.mainThread()).subscribe(onNext, onError, onComplete);
    }

    public <T> Disposable register(Class<T> eventType, Consumer<T> onNext, Consumer onError) {
        return toObservable(eventType).observeOn(AndroidSchedulers.mainThread()).subscribe(onNext, onError);
    }
}

还有一个通用的Event.java
依据自己的需要制定即可

/**
 * If you have any questions, you can contact by email {wangzhumoo@gmail.com}
 * @author 王诛魔 2017/10/19 下午5:38
 *
 * 发送事件包装类
 */
public class Event<T> {


    public static final int EVENT_CLOSE_ALL_ACTIVITY = 10001;

    /**
     * reserved data
     */
    private T data;

    /**
     * this code distinguish between different events
     */
    private int eventCode = -1;

    public Event(int eventCode) {
        this(eventCode, null);
    }

    public Event(int eventCode, T data) {
        this.eventCode = eventCode;
        this.data = data;
    }

    /**
     * get event code
     *
     * @return
     */
    public int getCode() {
        return this.eventCode;
    }

    /**
     * get event reserved data
     *
     * @return
     */
    public T getData() {
        return this.data;
    }


}

RxBus的使用

注册

最好是再基类中,加上.

if (useEventBus()) {
            //注册eventbus
            Disposable disposable = RxBus.getDefault()
                    .register(Event.class, new Consumer<Event>() {
                        @Override
                        public void accept(Event event)  {
                            int eventCode = event.getCode();
                            Log.e("RxBus", event.toString());
                            switch (eventCode) {
                                case Event.EVENT_CLOSE_ALL_ACTIVITY:

                                    break;
                                default:
                                    onEvent(event);
                                    break;
                            }
                        }
                    });
            addDispose(disposable);
}

其中:

• useEventBus()
  由子类实现,是否开启RxBus,避免浪费

• addDispose(disposable);
  addDispose 是基类的方法

    /**
     * RxJava 添加订阅
     */
    protected void addDispose(Disposable disposable) {
        if (mDisposables == null) {
            mDisposables = new CompositeDisposable();
        }
        //将所有disposable放入,集中处理
        mDisposables.add(disposable);
    }

统一管理,销毁时一起销毁,避免爆内存

• onEvent(event);
  是一个空方法,如果子类重写就把事件交给子类处理

发送

RxBus.getDefault().post(new Event(100,"message"))

//code最好在Event.java中统一定义,方便管理

处理

override fun onEvent(event: Event<*>?) {
    super.onEvent(event)
        
}

这一部分,其实在 注册 中已经完成,只要子类自己重写方法,就能取到数据