RxJava事件总线

Question

使用的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

Answer 1

在RxJava2的Action1已更名为Consumer的最短途径。

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

见What's different in 2.0

Answer 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); 
    } 
}