[Android]基於RxJava、RxAndroid的EventBus實現

以下內容為原創，歡迎轉載，轉載請注明

來自天天博客：http://www.cnblogs.com/tiantianbyconan/p/4578699.html 

 

Github：https://github.com/wangjiegulu/RxAndroidEventsSample

EventBus的作用是發布/訂閱事件總線，因為項目中用到RxJava、RxAndroid，所以完全可以使用RxJava、RxAndroid來實現EventBus。

 

1. 編寫RxBus，用於存儲所有事件Subjects。

事件是傳遞的最小單位，可以把任何類作為一個事件。

RxBus代碼如下：

/**
 * Author: wangjie
 * Email: [e-mail]
 * Date: 6/11/15.
 */
public class RxBus {
    private static final String TAG = RxBus.class.getSimpleName();
    private static RxBus instance;
    public static boolean DEBUG = false;

    public static synchronized RxBus get() {
        if (null == instance) {
            instance = new RxBus();
        }
        return instance;
    }

    private RxBus() {
    }

    private ConcurrentHashMap<Object, List<Subject>> subjectMapper = new ConcurrentHashMap<>();

    @SuppressWarnings("unchecked")
    public <T> Observable<T> register(@NonNull Object tag, @NonNull Class<T> clazz) {
        List<Subject> subjectList = subjectMapper.get(tag);
        if (null == subjectList) {
            subjectList = new ArrayList<>();
            subjectMapper.put(tag, subjectList);
        }

        Subject<T, T> subject;
        subjectList.add(subject = PublishSubject.create());
        if (DEBUG) Log.d(TAG, "[register]subjectMapper: " + subjectMapper);
        return subject;
    }

    public void unregister(@NonNull Object tag, @NonNull Observable observable) {
        List<Subject> subjects = subjectMapper.get(tag);
        if (null != subjects) {
            subjects.remove((Subject) observable);
            if (ABTextUtil.isEmpty(subjects)) {
                subjectMapper.remove(tag);
            }
        }

        if (DEBUG) Log.d(TAG, "[unregister]subjectMapper: " + subjectMapper);
    }

    public void post(@NonNull Object content) {
        post(content.getClass().getName(), content);
    }

    @SuppressWarnings("unchecked")
    public void post(@NonNull Object tag, @NonNull Object content) {
        List<Subject> subjectList = subjectMapper.get(tag);

        if (!ABTextUtil.isEmpty(subjectList)) {
            for (Subject subject : subjectList) {
                subject.onNext(content);
            }
        }
        if (DEBUG) Log.d(TAG, "[send]subjectMapper: " + subjectMapper);
    }
}

如上述代碼，RxBus只提供了register、unregister、post三個方法。

這里又加入了一個tag的概念，也可以理解為channel，注冊Subject、反注冊Subject和post事件的時候都需要這個tag，只有tag一致才能正常接收到事件。

比如有一個事件類HelloEvent，這個事件的作用是接收到后toast一個提示“hello”，如果兩個Activity都注冊了這個HelloEvent事件，但是沒有tag去限制，一旦post了一個helloEvent事件后，兩個Activity都會收到這個事件，導致兩個Activity都會toast。如果使用tag，post這個HelloEvent的時候可以設置這個tag，只有register時也使用了這個tag才會接收到這個event。

2. 在Present（如Activity的onCreate）中注冊一個Observer（以下以發送一個String類型的事件為例）

Observable<String> addOb = RxBus.get()
                .register("addFeedTag", String.class);

addOb.observeOn(AndroidSchedulers.mainThread())
                .subscribe(s -> {
                    // todo: Accept event and process here
                });

如上，注冊了一個String類型的事件，事件的tag是“addFeedTag”，用來增加一個Feed。使用RxAndroid在Action1中處理接收到的這個事件。

3. 在任何地方發送一個事件：

RxBus.get().post("addFeedTag", "hello RxBus!");

這里發送了一個tag為“addFeedTag”的String類型的事件。

4. 反注冊Observer：

RxBus.get().unregister("addFeedTag", addOb);

注意：這里的Tag都為“addFeedTag”。

 

下面使用注解的方式更簡單方便地使用RxBus（嗯－。－這里才是重點）。

首先來看下使用注解后的代碼：

1. 注冊Observer

這一步可以省略掉。

2. 發送一個事件（這里我們換一個事件：FeedItemClickEvent，我們定義這個事件是用來處理當Feed被點擊后的事件）

RxBus.get().post(new FeedItemClickEvent().setPosition(position).setFeed(feed));

3. 接收事件，然后處理

@Accept
public void onPostAccept(Object tag, FeedItemClickEvent event) {
　　Logger.d(TAG, "onPostAccept event: " + event);
　　Feed feed = event.getFeed();
　　// 跳轉到feed詳情頁面...
}

如上，這里只需要編寫一個方法，加上Accept注解，然后在方法中進行事件處理即可。

注意：方法名可以任意

方法參數一：必須為Object類型的tag；

方法參數二，如果這個方法只接收一種事件，則寫明具體的事件類型，如上；如果這個方法接收多種事件，則類型需要為Object。

4. 反注冊Observer

這一步也可以省略掉。

 

接收多種事件：

@Accept(
            acceptScheduler = AcceptScheduler.NEW_THREAD,
            value = {
                    @AcceptType(tag = ActionEvent.CLOSE, clazz = String.class),
                    @AcceptType(tag = ActionEvent.BACK, clazz = String.class),
                    @AcceptType(tag = ActionEvent.EDIT, clazz = String.class),
                    @AcceptType(tag = ActionEvent.REFRESH, clazz = String.class)
            }
    )
    public void onPostAccept(Object tag, Object actionEvent) {
        Logger.d(TAG, "[ActionEvent]onPostAccept action event name: " + actionEvent);
        // todo: Accept event and process here (in new thread)
    }

這里@Accept注解中設置了acceptScheduler為AcceptScheduler.NEW_THREAD，指明方法運行在子線程中.

value中指明了接收的事件類型，這里表示這個方法接收4種類型的事件：CLOSE, BACK, EDIT, REFRESH.

 

注解解釋：

@Accept注解

acceptScheduler: 指定被注解的方法運行的Scheduler。

value[]: AcceptType注解數組，用於指定接收事件的tag和class。

 

@AcceptType注解：

tag: 接收事件的tag
clazz: 接收事件的類型

 

AcceptScheduler：

詳情見：rx.schedulers.Schedulers和rx.android.schedulers.AndroidSchedulers

如果設置的是AcceptScheduler.EXECUTOR或AcceptScheduler.HANDLER，則需要在Application中配置Executor和Handler：

/**
 * Author: wangjie
 * Email: [e-mail]
 * Date: 6/15/15.
 */
public class MyApplication extends Application {
    private Executor acceptExecutor = Executors.newCachedThreadPool();
    private Handler handler = new Handler(Looper.getMainLooper());

    @Override
    public void onCreate() {
        super.onCreate();
        RxBus.DEBUG = true;

        DefaultAcceptConfiguration.getInstance().registerAcceptConfiguration(new DefaultAcceptConfiguration.OnDefaultAcceptConfiguration() {
            @Override
            public Executor applyAcceptExecutor() {
                return acceptExecutor;
            }

            @Override
            public Handler applyAcceptHandler() {
                return handler;
            }
        });
    }
}

因為需要對Accept和AcceptType注解的解析，所以項目的BaseActivity需要使用AIAppCompatActivity，然后實現parserMethodAnnotations()方法，使用RxBusAnnotationManager對注解進行解析。

 

參考：http://nerds.weddingpartyapp.com/tech/2014/12/24/implementing-an-event-bus-with-rxjava-rxbus/