制作簡單的RxBus實現(xiàn)事件總線

自從接觸了Rxjava之后就一直在使用Rxjava的方式來修改以前寫過的一些代碼，前段時間實現(xiàn)了Retrofit2+Rxjava的方式來請求數(shù)據(jù)感覺還不錯，那么既然使用了這種方式來獲取數(shù)據(jù)，自然也可以使用rx的方式來對數(shù)據(jù)進行處理。

之前我們可能使用callback、handler、Otto、EventBus...的方式處理數(shù)據(jù)，現(xiàn)在我們就來看下怎樣自己制作簡單的rxbus實現(xiàn)對數(shù)據(jù)的處理。

備注：rxjava已經(jīng)升級為2.0下面是升級為2.0的RxBus傳送門
http://www.itdecent.cn/p/b22e6b10c6cf

1 使用方式

1.1 注冊與取消注冊

在合適的地方進行注冊和取消注冊
RxBus.getInstance().register(this);
RxBus.getInstance().unRegister(this);

@Override
protected void onCreate(@Nullable Bundle savedInstanceState) {    
 super.onCreate(savedInstanceState);   
 RxBus.getInstance().register(this);
}

@Override
protected void onDestroy() {    
  super.onDestroy();    
  RxBus.getInstance().unRegister(this);
}

1.2 發(fā)布事件與訂閱事件

//發(fā)布
RxBus.getInstance().post(0, s.getBookses());

//訂閱
@Subscribe(tag = 0, thread = EventThread.MAIN_THREAD)
private void dataBinding(ArrayList<TravelNoteBook.Books> bookses) {    
...//處理邏輯
}

當(dāng)數(shù)據(jù)被發(fā)布之后，如果有訂閱者訂閱，就會獲取到發(fā)布過的數(shù)據(jù)。這也是一個最簡單的rxbus，那么我們就來看一下rxbus背后的處理，后面的內(nèi)容最好對rxjava和lambda方面的知識有一些了解。

2 實現(xiàn)最簡單的rxbus

對于上面用注解方式來訂閱事件的方式,我們先用最簡單的方式實現(xiàn)，然后把注解方式推導(dǎo)出來

2.1 發(fā)布與訂閱

首先我們要有一個Subject 來處理發(fā)布和訂閱事件
private final Subject bus;

/**
 * PublishSubject 創(chuàng)建一個可以在訂閱之后把數(shù)據(jù)傳輸給訂閱者Subject
 * SerializedSubject 序列化Subject為線程安全的Subject
 */
public RxBus() {
    bus = new SerializedSubject<>(PublishSubject.create());
}
然后我們來發(fā)布
public void post(int code, Object obj) {
    bus.onNext(obj);
}
最后訂閱
public <T> Observable tObservable(final Class<T> eventType) {
    return bus.ofType(eventType.getClass());//判斷接收事件類型
}
使用
APIServiceManager.getInstance()
        .getTravelNotesAPI()
        //Retrofit發(fā)送請求并返回數(shù)據(jù)
        .getTravelNotesList(query, page + "")
        //生命周期處理
        .compose(bindToLifecycle())
        //img01
        .compose(RxSchedulersHelper.io_main())
        .subscribe(s -> {
                    //發(fā)布
                    RxBus.getInstance().post(s.getBookses());
                });
//訂閱
Subscription subscription=RxBus.getInstance().tObservable(POJO.class)
        .subscribe(s -> {//事件處理});
//取消訂閱
subscription.unsubscribe();
img01

現(xiàn)在最原始的rxbus就成功了，現(xiàn)在可以進行簡單的事件的發(fā)布與訂閱，雖然可以進行發(fā)布于訂閱，但是在項目中還沒有辦法滿足我們的需求，那么下面我們就需要對現(xiàn)在的這種方式進行優(yōu)化。

2.2 rxbus優(yōu)化

先看一下優(yōu)化后rxbus包的內(nèi)容


Paste_Image.png

Subscribe //訂閱注解
EventTag //事件標(biāo)示
EventThread //事件線程
Msg //發(fā)布事件時默認封裝pojo
RxBus //rxbus主要處理

一共五個類主要的就是rxbus，那我們就先看一下rxbus中到底做了哪些處理

2.2.1 注冊
然后需要一個map空間存放訂閱者方便取消注冊
//存放訂閱者信息
private Map<Object, List<Subscription>> subscriptions = new HashMap<>();
現(xiàn)在來處理注冊事件

下面代碼核心思想是，獲取Object 中被Subscribe注解的方法并通過反射的方式，在數(shù)據(jù)被發(fā)布后通過invoke調(diào)用方法，并在訂閱后把訂閱者加入map空間方便解除注冊。

public void register(Object subscriber) {
  Observable.just(subscriber)
  //判斷訂閱者不為空            
  .filter(s -> s != null)
  .map(s -> s.getClass())            
  //獲取訂閱者方法并且用Observable裝載            
  .flatMap(s -> Observable.from(s.getDeclaredMethods()))
  //方法必須被Subscribe注解            
  .filter(m -> m.isAnnotationPresent(Subscribe.class))
  //使非public方法可以被invoke            
  .doOnNext(m -> m.setAccessible(true))
  .subscribe(m -> {                
    addSubscription(m,subscriber);            
  });
}

private void addSubscription(Method m,Object subscriber){
//獲取方法內(nèi)參數(shù)    
Class[] parameterType = m.getParameterTypes();    
//只獲取第一個方法參數(shù)，否則默認為Object    
Class cla = Object.class;    
if (parameterType.length > 1) {
    cla = parameterType[0];
}
//獲取注解
Subscribe sub = m.getAnnotation(Subscribe.class);    
//訂閱事件    
Subscription subscription = tObservable(sub.tag(), cla)
.observeOn(EventThread.getScheduler(sub.thread()))            
.subscribe(o -> {                        
  try {                            
    m.invoke(subscriber, o);                        
  } catch (IllegalAccessException e) {  
    e.printStackTrace();                        
  } catch (InvocationTargetException e) {  
    e.printStackTrace();                       
  }                    
}, e -> System.out.println("this object is not invoke"));  
  putSubscriptionsData(subscriber,subscription);
}

/**
 * 添加訂閱者到map空間來unRegister
 * @param subscriber 訂閱者
 * @param subscription 訂閱者 Subscription
 */
private void putSubscriptionsData(Object subscriber,Subscription subscription){
    List<Subscription> subs = subscriptions.get(subscriber);
    if (subs == null) {
        subs = new ArrayList<>();
    }
    subs.add(subscription);
    subscriptions.put(subscriber, subs);
}

代碼中我們調(diào)用過這么一個方法：tObservable(sub.tag(), cla)
這個就是剛才的tObservable方法，然后增加了一個int類型的參數(shù)用于標(biāo)示在同一個Object下使用不同的事件處理方法

/**
 * 訂閱事件
 * @return
 */
public <T> Observable tObservable(int code, final Class<T> eventType) {
    return bus.ofType(Msg.class)//判斷接收事件類型
            .filter(new Func1<Msg, Boolean>() {
                @Override
                public Boolean call(Msg o) {
                    //過濾code相同的事件
                    return o.code == code;
                }
            })
            .map(new Func1<Msg, Object>() {
                @Override
                public Object call(Msg o) {
                    return o.object;
                }
            })
            .cast(eventType);
}

對于post也相應(yīng)的增加了int參數(shù)，通過對code的判斷bus會返回eventType類型的Observable數(shù)據(jù)，并傳入備注解的相應(yīng)方法中。這樣就完成了整個事件的注冊過程

2.2.2 發(fā)布數(shù)據(jù)

把數(shù)據(jù)包裝以下并發(fā)送

public void post(int code, Object obj) {
    bus.onNext(new Msg(code, obj));
}
2.2.3 取消注冊
public void unRegister(Object subscriber) {
    if (subscriber == null) {
        return;
    }
    List<Subscription> subs = subscriptions.get(subscriber);
    if (subs != null) {
        for (Subscription sub : subs) {
            if (sub != null)
                sub.unsubscribe();
        }
        subscriptions.remove(subscriber);
    }
}

取消注冊主要就是判斷map中相同的subscriber，然后調(diào)用unsubscribe()取消事件訂閱并刪除相應(yīng)的map數(shù)據(jù)，保證訂閱事件與activity生命周期相同防止內(nèi)存溢出。

使用
APIServiceManager.getInstance()
        .getTravelNotesAPI()
        .getTravelNotesList(query, page + "")
        .compose(bindToLifecycle())
        .compose(RxSchedulersHelper.io_main())
        //對返回數(shù)據(jù)的優(yōu)先處理，如果數(shù)據(jù)有異常則會執(zhí)行error
        .compose(SchedulersHelper.handleResult())
        .subscribe(s -> {
                    //發(fā)布
                    RxBus.getInstance().post(0, s.getBookses());
                },
                e -> {
                    //發(fā)布（error）
                    RxBus.getInstance().post(RxBus.TAG_ERROR, e.getMessage());
                });

@Subscribe(tag = 0, thread = EventThread.MAIN_THREAD)
private void dataBinding(ArrayList<TravelNoteBook.Books> bookses) {
    //事件處理
}
@Subscribe(tag =RxBus.TAG_ERROR)
private void dataError(String error) {
    ToastUtil.getInstance().makeShortToast(this, error);
}

3 總結(jié)

現(xiàn)在優(yōu)化過后最簡單的rxbus事件總線就已經(jīng)實現(xiàn)，可以對數(shù)據(jù)進行發(fā)布和訂閱處理，相關(guān)的代碼和demo可以訪問我的git:
https://github.com/hackerlc/GearApplication
或者在gradle中直接引用：

compile 'com.joker.gear:com-joker-gear:1.4.1'

當(dāng)然因為現(xiàn)在rxbus是最簡單的版本，不論是代碼還是實現(xiàn)方式還有很多不足的地方，在實際項目中也會遇到更多的問題，這里只提供了只用rxbus的實現(xiàn)思路，隨著rxbus的不斷完善相信后面會有更加健壯的代碼呈現(xiàn)出來。

4 相關(guān)引用

在實現(xiàn)rxbus上看了很多的文章和git，也非常感謝這些作者的分享，當(dāng)然如果在文章引用過程中有什么問題，請聯(lián)系我我會作出修改，以下是相關(guān)的引用連接。
http://www.open-open.com/lib/view/open1444011371541.html
http://www.itdecent.cn/p/ca090f6e2fe2
https://github.com/AndroidKnife/RxBus
https://github.com/1030310877/JoeRxBus
https://github.com/ch331917692/RxBus
http://www.itdecent.cn/p/f3f0eccbcd6f

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容