前言

??Android開發(fā)中，我們使用MVP模式開發(fā)的時候，涉及到Model層和Presenter層的通信的時候，我們一般都需要使用Handler 來實現(xiàn)，因為Model層我們一般處理具體的業(yè)務邏輯，不可避免的需要進行進行網(wǎng)絡請求，由于Android中是不允許在UI線程中進行網(wǎng)絡請求的，而更新UI必須是在UI線程中進行，那么就涉及到了線程的切換，在Rxjava 出來之前，我們都是通過Handler來實現(xiàn)，導致需要在Presenter層實現(xiàn)Handler并且要給Model層，而且處理回掉都是在Handler的handMessage()方法中進行，導致方法中維護麻煩，容易翻車。

??在Rxjava出現(xiàn)過后，在Model層我們可以使用Rxjava來切換線程，這樣做的確可以不用Handler了，但是還是在與Presenter的交互過程了還是需要接口，這樣做問題就是Presenter 中調(diào)用Model層方法的時候需要傳遞接口來實現(xiàn)在結束網(wǎng)絡請求過后，立即通知并回傳結果給Presenter。如果一個Model中的網(wǎng)絡請求非常多，那么需要定義非常多的回掉接口。這樣Module 和 Presenter也是高度耦合。不利于維護。
首先需要定義一個注解接口

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RegisterBus {
}

實現(xiàn)RxBus類

package com.example.baselib.http.databus;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;

import rx.Observable;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.schedulers.Schedulers;

/**
 * Created by huangjie on 2018/12/24.
 */

public class RxBus {
    /**
     * 單例對象
     */
    private static volatile RxBus instance;

    /**
     * 觀察者集合
     */
    private static Set<Object> subscribers;

    private RxBus() {
        subscribers = new CopyOnWriteArraySet<>();
    }

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

    /**
     * 添加訂閱
     *
     * @param object
     */
    public void registSubscribe(Object object) {
        subscribers.add(object);
    }

    /**
     * 接觸訂閱
     *
     * @param object
     */
    public void unRegistSubscribe(Object object) {
        subscribers.remove(object);
    }

    /**
     * 轉換
     *
     * @param func1
     */
    public void Process(Func1 func1) {
        Observable.just("")
                .subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .map(func1)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1() {
                    @Override
                    public void call(Object data) {
                        if (data == null) {
                            return;
                        }
                        callRegister(data);
                    }
                });
    }

    /**
     * @param data
     */
    private void callRegister(Object data) {
        for (Object target : subscribers) {
            callMethodByAnnotiation(target, data);
        }
    }

    /**
     * 通過注解找到方法 然后通過反射調(diào)用方法
     *
     * @param target
     * @param data
     */
    public void callMethodByAnnotiation(Object target, Object data) {
        Method[] methodArray = target.getClass().getDeclaredMethods();
        for (int i = 0; i < methodArray.length; i++) {
            Method method = methodArray[i];
            if (method.isAnnotationPresent(RegisterBus.class)) {
                Class<?> paramType = method.getParameterTypes()[0];
                if (data.getClass().getName().equals(paramType.getName())) {
                    try {
                        method.invoke(target, new Object[]{data});
                    } catch (IllegalAccessException e) {
                        e.printStackTrace();
                    } catch (InvocationTargetException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }

}

核心思想就是通過Rxjava的可以通過map來實現(xiàn)流的轉換,如果不熟悉Rxjava的map用法，可以去看給Android 開發(fā)者的 RxJava 詳解這篇文章
使用demo:

package com.example.huangjie.mytaxi;

import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;
import android.widget.Toast;
import com.example.baselib.http.IResponse;
import com.example.baselib.http.databus.RegisterBus;
import com.example.baselib.http.databus.RxBus;
import com.example.baselib.http.impl.BaseRequest;
import com.example.baselib.http.impl.HttpClientManagerImpl;
import rx.functions.Func1;
public class MainActivity extends AppCompatActivity {

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

    private void testRxBus() {
        RxBus.getInstance().registSubscribe(this);
        RxBus.getInstance().Process(new Func1() {
            @Override
            public Object call(Object o) {
                BaseRequest request = new BaseRequest("http://www.baidu.com");
                HttpClientManagerImpl clientManager = new HttpClientManagerImpl();
                IResponse response = clientManager.get(request, false);
                String data = response.getData();
                Log.e("huangjie", "http 請求的結果 " + data);
                return data;
            }
        });
    }

    @RegisterBus
    public void callResultMethod(String data) {
        Toast.makeText(this, data, Toast.LENGTH_SHORT).show();
    }
    @Override
    protected void onDestroy() {
        RxBus.getInstance().unRegistSubscribe(this);
        super.onDestroy();
    }
}

使用demo下載地址