RxJava 事件总线答案

【问题标题】：RxJava Event BusRxJava 事件总线
【发布时间】：2017-05-03 13:48:10
【问题描述】：

使用RxJava 和RxAndroid 的第一个版本，我将class 视为EventBus：

public class RxBus {
private static RxBus instance;
private PublishSubject<Object> subject = PublishSubject.create();

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

public void setMessage(Object object) {
    subject.onNext(object);
}

public Observable<Object> getEvents() {
    return subject;
}
}

在任何类中通过instanceOf 获取实例，我使用setMessage 方法发出消息并使用以下代码获取发出的消息：

  bus.getEvents().subscribe(new Action1<Object>() {
        @Override
        public void call(Object o) {
            if (o instanceof String) {
                //TODO
            }
        }
    });

Action1 来自 rx.functions 包。尝试迁移使用 RxJava 2 我无法导入它。

请告诉我，将RxJava 2 用作EventBus 的最短方法是什么

【问题讨论】：

标签： android rx-java rx-android rx-java2

【解决方案1】：

在 RxJava2 中，Action1 已重命名为 Consumer。

其余的操作接口根据 Java 8 功能类型命名。无参数 Action0 被 io.reactivex.functions.Action 替换为运算符，java.lang.Runnable 替换为 Scheduler 方法。 Action1 已重命名为 Consumer 并且 Action2 被称为 BiConsumer。 ActionN 被 Consumer<Object[]> 类型声明替换。

见What's different in 2.0

【讨论】：

【解决方案2】：

这是 RxJava2 中事件总线的一个很好的实现（代码复制自 gist）

import io.reactivex.Observable;
import io.reactivex.subjects.PublishSubject;

public class RxBus {
    private static volatile RxBus sRxBus = null;
    private PublishSubject<Object> mPublishSubject = PublishSubject.create();

    private RxBus() {
    }

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

    public <T> Observable<T> subscribe(Class<T> cls) {
        return mPublishSubject
                .filter(o -> o.getClass().equals(cls))
                .map(o -> (T) o);
    }

    public void post(Object obj) {
        mPublishSubject.onNext(obj);
    }
}

【讨论】：