前言

  Android开发中，我们使用MVP模式开发的时候，涉及到Model层和Presenter层的通信的时候，我们一般都需要使用Handler 来实现，因为Model层我们一般处理具体的业务逻辑，不可避免的需要进行进行网络请求，由于Android中是不允许在UI线程中进行网络请求的，而更新UI必须是在UI线程中进行，那么就涉及到了线程的切换，在Rxjava 出来之前，我们都是通过Handler来实现，导致需要在Presenter层实现Handler并且要给Model层，而且处理回掉都是在Handler的handMessage()方法中进行，导致方法中维护麻烦，容易翻车。

  在Rxjava出现过后，在Model层我们可以使用Rxjava来切换线程，这样做的确可以不用Handler了，但是还是在与Presenter的交互过程了还是需要接口，这样做问题就是Presenter 中调用Model层方法的时候需要传递接口来实现在结束网络请求过后，立即通知并回传结果给Presenter。如果一个Model中的网络请求非常多，那么需要定义非常多的回掉接口。这样Module 和 Presenter也是高度耦合。不利于维护。
首先需要定义一个注解接口

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

实现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);
        }
    }

    /**
     * 通过注解找到方法 然后通过反射调用方法
     *
     * @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来实现流的转换,如果不熟悉Rxjava的map用法，可以去看给Android 开发者的 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下载地址