基于RxJava的事件总线RxBus

RxJava2.0的简单使用
RxJava2.0---创建被观察者
基于RxJava的事件总线RxBus

RxBus是基于 RxJava 的一种事件总线，相信许多人都了解一些，在RxBus出来之前，相信更多的人是使用EventBus，因为现在 RxJava 太火了，用它几行代码就可以写出事件总线,所以要做到与时俱进。熟悉RxJava的人应该都知道RxJava从1.0版本升级到2.0，做了很多优化的地方 What’s different in 2.0 。下面总结RxJava1.0和 RxJava2.0 版本的 RxBus 写法。

RxJava1.0版本的RxBus

public class RxBus {
private static volatile RxBus defaultInstance;

private final Subject<Object, Object> bus;
// PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者
public RxBus() {
  bus = new SerializedSubject<>(PublishSubject.create());
}
// 单例RxBus
public static RxBus getDefault() {
    if (defaultInstance == null) {
        synchronized (RxBus.class) {
            if (defaultInstance == null) {
                defaultInstance = new RxBus();
            }
        }
    }
    return defaultInstance ;
}
// 发送一个新的事件
public void post (Object o) {
    bus.onNext(o);
}
// 根据传递的 eventType 类型返回特定类型(eventType)的 被观察者
public <T> Observable<T> toObservable (Class<T> eventType) {
    return bus.ofType(eventType);
}

 public Observable<Object> toObserverable() {
    return bus;
    }
}

RxJava2.0版本的RxBus

在RxJava2.0中把背压和非背压处理分开了，所有出现两种模式。

1. 没有背压处理（Backpressure）的 Rxbus

2. 有背压处理的 RxBus

    public class RxBus {
   
    //    // 有背压处理的 RxBus
    //    private final FlowableProcessor<Object> bus;
   
    //非背压处理
    private final Subject<Object> bus;
   private static volatile RxBus defaultRxBus;
   
   private RxBus() {
    //非背压处理
    bus = PublishSubject.create().toSerialized();
   
    //        // 有背压处理的 RxBus
    //        bus = PublishProcessor.create().toSerialized();
   
    }
   
    public static RxBus getInstance() {
    if (null == defaultRxBus) {
        synchronized (RxBus.class) {
            if (null == defaultRxBus) {
                defaultRxBus = new RxBus();
            }
        }
    }
    return defaultRxBus;
    }
   
    public void post(Object o) {
    bus.onNext(o);
    }
   
    public boolean hasObservable() {
    return bus.hasObservers();
    }
   
    public Observable<Object> toObservable() {
    return bus;
    }
   
    /*
     * 转换为特定类型的Obserbale
     */
    public <T> Observable<T> toObservable(Class<T> type) {
    return bus.ofType(type);
    }
    }
   

发送通知

 RxBus.getInstance().post(new byte[]{0x01, 0x02, 0x03, 0x04});

监听通知

 subscribe = RxBus.getInstance().toObservable().subscribe(new Consumer<Object>() {
        @Override
        public void accept(@NonNull Object o) throws Exception {
            if (o instanceof String) {
                Toast.makeText(MainActivity.this, "o:" + o, Toast.LENGTH_SHORT).show();
            } else if (o instanceof byte[]) {
                Toast.makeText(MainActivity.this, "length:" + ((byte[]) o).length, Toast.LENGTH_SHORT).show();
            }
        }
    }, new Consumer<Throwable>() {
        @Override
        public void accept(@NonNull Throwable throwable) throws Exception {
            Toast.makeText(MainActivity.this, throwable.getLocalizedMessage(), Toast.LENGTH_SHORT).show();
        }
    });

RXBus的基本使用就这么简单，O了。