RxBus在rxjava2.0 的使用

时间:2023-01-20 17:44:56

   今天自己刚好要写一个例子,本来想要用evenbus,但是想到自己用的是rxjava,在用evenbus就比较不和了吧! 所以就自己写了个rxbus

   1. 首先是gradle的配置

compile 'io.reactivex.rxjava2:rxjava:2.0.8'
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile 'com.jakewharton.retrofit:retrofit2-rxjava2-adapter:1.0.0'
compile 'com.jakewharton.rxrelay2:rxrelay:2.0.0'//出现异常也不会终止订阅关系 不使用第一个rxbus的写法的话,可以不需要这个
 2.RxBus的写法
 1. 在订阅者处理事件出现异常后,订阅者无法再收到事件,这是 RxJava 当初本身的设计原则,但是在事件总线中这反而是个问题,不过
 JakeWharton 大神写了即使出现异常也不会终止订阅关系的 RxRelay,所以基于 RxRelay 就能写出有异常处理能力的 Rxbus。
import com.jakewharton.rxrelay2.PublishRelay;
import com.jakewharton.rxrelay2.Relay;

import io.reactivex.Observable;

/**
* 有异常处理的 Rxbus
* Created by huxh on 2017/4/10.
*/

public class RxBus {
private static volatile RxBus instance;
private final Relay<Object> mBus;

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

public static RxBus getDefault() {
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();
}

}
2.没有背压处理(Backpressure)的 Rxbus
import io.reactivex.Observable;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.Subject;

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();
}
}

3.有背压处理的 RxBus

import io.reactivex.Flowable;
import io.reactivex.processors.FlowableProcessor;
import io.reactivex.processors.PublishProcessor;

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();
}
}
有些朋友不知道有背压与没有背压是什么意识,可以参考 http://flyou.ren/2017/04/05/关于RxJava背压/讲的很详细

好了,到了如何使用rxbus了,其实很简单

比如A跳转到B ,我这边是recycleview 的item点击跳转,传一个对象
   @Override
public void onItemClick(View view, int position) {
ActicleBean.OthersBean bean = mDataList.get(position);
RxBus.getDefault().post(bean);
startActivity(new Intent(ArticleActivity.this,ActicleListActivity.class));
}

在B类上 直接写就可以获得了你所需要传的对象

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;

import com.frame.huxh.mvpdemo.R;
import com.frame.huxh.mvpdemo.bean.ActicleBean;
import com.frame.huxh.mvpdemo.rxbus.RxBus;
import com.frame.huxh.mvpdemo.utils.ToastUtils;

import io.reactivex.annotations.NonNull;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;

public class ActicleListActivity extends AppCompatActivity {

@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_acticle_list);
operateBus();
}

/**
* RxBus
*/
private void operateBus() {
RxBus.getDefault().toObservable()
.map(new Function<Object, ActicleBean.OthersBean>() {
@Override
public ActicleBean.OthersBean apply(@NonNull Object o) throws Exception {
return (ActicleBean.OthersBean) o;
}

})
.subscribe(new Consumer<ActicleBean.OthersBean>() {
@Override
public void accept(@NonNull ActicleBean.OthersBean othersBean) throws Exception {
if (othersBean != null) {
ToastUtils.toast(ActicleListActivity.this, "othersBean" + othersBean.getDescription());
}
}

});
}
}