jeremyliao/liveeventbus

:mailbox_with_mail:EventBus for Android，消息总线，基于LiveData，具有生命周期感知能力，支持Sticky，支持AndroidX，支持跨进程，支持跨APP

LiveEventBus

LiveEventBus是一款Android消息总线，基于LiveData，具有生命周期感知能力，支持Sticky，支持AndroidX，支持跨进程，支持跨APP

为什么要用LiveEventBus

生命周期感知

• 消息随时订阅，自动取消订阅
• 告别消息总线造成的内存泄漏
• 告别生命周期造成的崩溃

范围全覆盖的消息总线解决方案

• 进程内消息发送
• App内，跨进程消息发送
• App之间的消息发送

更多特性支持

• 免配置直接使用，懒人最爱
• 支持Sticky粘性消息
• 支持AndroidX
• 支持延迟发送
• 观察者的多种接收模式（全生命周期/激活状态可接受消息）

常用消息总线对比

消息总线	延迟发送	有序接收消息	Sticky	生命周期感知	跨进程/APP	线程分发
EventBus	❌	✅	✅	❌	❌	✅
RxBus	❌	❌	✅	❌	❌	✅
LiveEventBus	✅	✅	✅	✅	✅	❌

想了解更多？请点击：全面了解Android消息总线

使用

1.8及以上版本全面迁移至maven，同时groupID变为io.github.jeremyliao，1.8以下版本保留JCenter

maven

• 非AndroidX

implementation 'io.github.jeremyliao:live-event-bus:1.8.0'

• AndroidX

implementation 'io.github.jeremyliao:live-event-bus-x:1.8.0'

JCenter

• 非AndroidX

implementation 'com.jeremyliao:live-event-bus:1.7.3'

• AndroidX

implementation 'com.jeremyliao:live-event-bus-x:1.7.3'

快速开始

订阅消息

• 以生命周期感知模式订阅消息

LiveEventBus
	.get("some_key", String.class)
	.observe(this, new Observer<String>() {
	    @Override
	    public void onChanged(@Nullable String s) {
	    }
	});

• 以Forever模式订阅消息

LiveEventBus
	.get("some_key", String.class)
	.observeForever(observer);

发送消息

• 不定义消息直接发送

LiveEventBus
	.get("some_key")
	.post(some_value);

• 先定义消息，再发送消息

public class DemoEvent implements LiveEvent {
    public final String content;

    public DemoEvent(String content) {
        this.content = content;
    }
}

LiveEventBus
        .get(DemoEvent.class)
        .post(new DemoEvent("Hello world"));

详细使用文档

获取Observable

通过name获取Observable

Observable<T> get(@NonNull String key, @NonNull Class<T> type)

Observable<Object> get(@NonNull String key)

通过event type获取Observable

<T extends LiveEvent> Observable<T> get(@NonNull Class<T> eventType)

消息发送

进程内发送消息

void post(T value)

App内发送消息，跨进程使用

void postAcrossProcess(T value)

App之间发送消息

void postAcrossApp(T value)

进程内发送消息，延迟发送

void postDelay(T value, long delay)

进程内发送消息，延迟发送，带生命周期

void postDelay(LifecycleOwner sender, T value, long delay)

进程内发送消息，有序发送

void postOrderly(T value)

以广播的形式发送一个消息

• 需要跨进程、跨APP发送消息的时候调用该方法
• 建议尽量使用postAcrossProcess、postAcrossApp

void broadcast(T value, boolean foreground, boolean onlyInApp)

消息订阅

以生命周期感知模式订阅消息

• 具有生命周期感知能力，LifecycleOwner销毁时自动取消订阅，不需要调用removeObserver

void observe(@NonNull LifecycleOwner owner, @NonNull Observer<T> observer)

以Forever模式订阅和取消订阅消息

• Forever模式订阅消息，需要调用removeObserver取消订阅

void observeForever(@NonNull Observer<T> observer)

取消订阅消息

void removeObserver(@NonNull Observer<T> observer)

Sticky模式订阅消息

• Sticky模式
• 支持在订阅消息的时候设置Sticky模式，这样订阅者可以接收到之前发送的消息。
• 以Sticky模式订阅消息，具有生命周期感知能力，LifecycleOwner销毁时自动取消订阅，不需要调用removeObserver

void observeSticky(@NonNull LifecycleOwner owner, @NonNull Observer<T> observer)

Sticky模式Forever订阅消息

• Forever模式订阅消息，需要调用removeObserver取消订阅，Sticky模式

void observeStickyForever(@NonNull Observer<T> observer)

跨进程消息

支持对基本数据类型消息的跨进程发送

1. int
2. float
3. long
4. boolean
5. double
6. String

支持Serializable和Parcelable类型消息的跨进程发送

• 提供SerializableProcessor
• 提供ParcelableProcessor

支持Bean类型消息的跨进程发送

• 提供GsonProcessor以Gson方式提供支持
• 需要用注解@IpcConfig指定GsonProcessor：

@IpcConfig(processor = GsonProcessor.class)

1.8及以上版本由于拆分了GsonProcessor，需要引入lebx-processor-gson

• 非AndroidX

implementation 'io.github.jeremyliao:leb-processor-gson:x.x.x'

• AndroidX

implementation 'io.github.jeremyliao:lebx-processor-gson:x.x.x'

支持自定义扩展

• 实现自定义Processor，实现Processor接口
• 用注解@IpcConfig指定自定义Processor

老版本文档

• 如果使用1.4.X版本，请参考使用方法
• 如果使用1.3.X及以下版本，请参考使用方法

更多使用场景

SmartEventBus

SmartEventBus是一个Android平台的消息总线框架，这是一款非常smart的消息总线框架，能让你定制自己的消息总线。 SmartEventBus

在组件化中使用LiveEventBus

android-modular

配置

在Application.onCreate方法中配置：

LiveEventBus
        .config()
        ...

• lifecycleObserverAlwaysActive 配置LifecycleObserver（如Activity）接收消息的模式（默认值true）

• autoClear 配置在没有Observer关联的时候是否自动清除LiveEvent以释放内存（默认值false）

更多配置信息，请点击：LiveEventBus的配置

混淆规则

-dontwarn com.jeremyliao.liveeventbus.**
-keep class com.jeremyliao.liveeventbus.** { *; }
-keep class android.arch.lifecycle.** { *; }
-keep class android.arch.core.** { *; }

for androidx:

-dontwarn com.jeremyliao.liveeventbus.**
-keep class com.jeremyliao.liveeventbus.** { *; }
-keep class androidx.lifecycle.** { *; }
-keep class androidx.arch.core.** { *; }

常见问题Q&A

Issues上最经常被提问的问题

• Q：收不到消息怎么办？

  A：通过Console.getInfo()获取即时的调试信息，主要去观察对应的key下面有没有你关注的Observer（Console的用法）

  *********Event info*********
  Event name: key_test_delay_life
      version: -1
      hasActiveObservers: true
      hasObservers: true
      ActiveCount: 1
      ObserverCount: 1
      Observers: 
          [com.jeremyliao.liveeventbus.core.LiveEventBusCore$ObserverWrapper@992681d=android.arch.lifecycle.ExternalLiveData$ExternalLifecycleBoundObserver@bc258f4]
  

• Q：收到重复的消息怎么办？

  A：同样通过Console.getInfo()获取即时的调试信息，主要去观察有没有重复注册的Observer（Console的用法）

• Q：JCenter要关闭了，什么时候迁移？

  A：1.8及以上版本全面迁移至maven，同时groupID变为io.github.jeremyliao，1.8以下版本保留JCenter

• Q：如何传递List或者Pair<Boolean, Double>这种泛型对象？

  A：代码上的写法有两种，供参考：

  LiveEventBus
          .get(KEY_TEST_OBSERVE_FOREVER, List.class)
          .observeForever(new Observer<List>() {
              @Override
              public void onChanged(@Nullable List list) {
                  List<String> stringList = list;
              }
          });
  

  Observable<List<String>> observable = LiveEventBus.get(KEY_TEST_OBSERVE_FOREVER);
  observable.observeForever(new Observer<List<String>>() {
      @Override
      public void onChanged(@Nullable List<String> strings) {
      }
  });
  

• Q：项目中不用Gson库，如何去掉依赖？

  A：使用1.8以上的版本，对gson相关代码进行了拆分

若问题不能解决，请提issue。

其他分支版本

AndroidX

• 支持AndroidX
• 同master版本一致

classic

• 经典实现版，整个实现就一个java文件
• 只支持激活状态（Started）可以实时收到消息，非激活状态（Stoped）无法实时收到消息，需等到Activity重新变成激活状态，方可收到消息
• 不支持跨进程通信

v2

• v2版，历史版本，已废弃
• 为了解决非激活态不能实时收到消息的问题，采用修改LiveData源码的方式实现

示例和DEMO

文档

实现原理

• LiveEventBus的实现原理可参见作者在美团技术博客上的博文： Android消息总线的演进之路：用LiveDataBus替代RxBus、EventBus
• 该博文是初版LiveEventBus的实现原理，与当前版本的实现可能不一致，仅供参考

质量

• 编写了30个测试用例以确保LiveEventBus能够正常运行。
• 具体测试用例参见LiveEventBusTest

版本

版本	功能
1.8.x	迁移至maven，拆分gson-converter
1.7.x	优化接口设计，优化实现逻辑，修复一些问题
1.6.x	优化接口设计，优化实现逻辑，修复一些问题
1.5.x	优化接口设计，使用起来更简洁
1.4.x	简化对外暴露的接口，重构核心实现，支持前后台线程调用
1.3.x	支持跨进程、跨APP通信
1.2.x	支持接收消息的模式，支持AndroidX
1.1.x	修复了一些问题
1.0.x	初版，支持基本功能

主要功能提交记录

1. 主要功能完成（Jul 11, 2018）
2. 支持Sticky（Aug 8, 2018）
3. 修复在后台线程PostValue会丢失消息的问题（Aug 9, 2018）
4. 解决发送给Stop状态Observer消息无法及时收到的问题（Aug 18, 2018）
5. 解决了Resumed状态的Activity发生订阅，订阅者会收到订阅之前发布的消息的问题。特别感谢@MelonWXD发现了这个问题（Dec 8，2018）
6. 在removeObserver的时候，检查livedata上有没有observer，没有则删除这个livadata，以减少内存占用。特别感谢@GreenhairTurtle提供的解决方案（Dec 27，2018）
7. 支持设置LifecycleObserver接收消息的模式，支持在整个生命周期实时接收消息和只在激活态实时接收消息两种模式（Jan 22，2019）
8. 支持AndroidX（Mar 8，2019）
9. 支持跨进程、跨APP（Mar 26，2019）
10. 简化对外暴露的接口，重构核心实现，支持前后台线程调用（Apr 4，2019）

其他

• 欢迎提Issue与作者交流
• 欢迎提Pull request，帮助 fix bug，增加新的feature，让LiveEventBus变得更强大、更好用

More Open Source by JeremyLiao

1. InterfaceLoader 史上最好用的Android跨进程接口调用框架件
2. FastSharedPreferences 一个Android平台的高性能key-value组件
3. SmartEventBus SmartEventBus是一个Android平台的消息总线框架，这是一款非常smart的消息总线框架，能让你定制自己的消息总线。
4. android-modular 一套Android组件化的实施方案和支撑框架
5. DataLoader 一个Android异步数据加载框架，用于Activity打开之前预加载数据，页面启动速度优化利器

更多，请点击

深入学习Android系列，让你精通Android

Flutter系列

• JeremyLiao/flutter-learn
• alibaba/flutter-go

Kotlin系列

• MindorksOpenSource/from-java-to-kotlin
• JeremyLiao/kotlin-compare-to-java

Gradle系列

• JeremyLiao/android-gradle-study

算法系列

• labuladong/fucking-algorithm

测试系列

• JeremyLiao/jacoco-android-demo

招聘|抖音成都招聘|有兴趣的同学可以点进去看一下，谢谢

liveeventbus's Issues

关于miniSdkVersion为15的问题

为什么miniSdkVersion 版本为15 而不是更低
集成1.2.0的时候 发现提示这个错误，我目前主项目是14 能不调高就不调了

the difference between assembleDebug and assembleRelease

when I use assembleDebug to make a apk ,it worked well ,but when i use assembleRelease,the apk not work(I add the proguard rules)

混淆怎么配置？

androidx 使用稳定版本

想使用androidx版本，结果发现是beta版
implementation "androidx.appcompat:appcompat:1.0.2"

readme 文档有误

文档配置那一处的说明有问题。与代码不符合

isCallOnObserve 这个函数是干什么用的？

希望支持发送空消息

并不是每次都需要 postValue(value)，有时候只需要一个 key 发送一个事件即可，希望可以支持一下 pos() 发送一个空消息

请问大佬的类头注释怎么写的

没错，就是那个很*气的LiveEventBus
一个个打出来的吗？
还是有啥便捷方法，我也想玩点花的

能否提供和主流 bus 的 性能对比 benchmark

用下来总体感受是简单好用，赞一个。
如果在工程里大规模应用还有个疑虑就是性能怎么样？
很多消息同时发送时是否会有延迟？
以及依赖了 MainLooper 的排队处理会不会因为主线程比较繁忙导致消息相应迟缓，这个现象在应用启动时已经发现了，会延迟 500ms-1s 相应 onChange

APP不杀死放入后台一段时间，重新打开的时候，无法监听到事件

在MainActivity中，当activity调用onResume的时候会post事件。在mainAcivity内的fragment中监听事件，如果我按home键回到主页，过了半个小时再打开的时候（此时进程没有被杀死，没有冷/热启动，可以瞬间回到mainActivity），此时会调用onResume方法，会post事件。然而fragment内却收不到事件。应该是监听失效了

关于去除反射，重构项目的一点思考

因为Android对反射加强管理了，版本兼容容易出现问题，所以一直不太希望代码中过多的使用反射去实现功能，所以这几天参考作者的思路对项目进行了重构，重构基于作者的两个项目进行了合并，如有好的建议欢迎大家一起讨论
https://github.com/codyer/LiveEventBus

同时感谢原作者的付出

在 androidx中用不了

当我迁移到androidx后，发现用不了了。LiveEventBus这个库真的很好用，我的很多项目都依赖他。希望作者大神能尽快支持androidx。感谢，您辛苦了！

无法实现自动解绑

借助LiveData的特性，可以在LifecycleOwner生命周期结束时自动解绑相应的Observer，但是内部维持的Map<String, LiveEvent<Object>> bus 好像没有把相应的key进行移除。目前好像是只有手动调用removeObserverInternal( Observer<T> observer)时才会尝试进行解绑。
可否重写LiveData的removeObserver方法，比如：

@Override
    public void removeObserver(@NonNull Observer<T> observer) {
        super.removeObserver(observer);
        if (!hasObservers()) {
            // set listener
        } 
    }

通过回调的方式进行一个自动解绑

为什么要打破系统帮我们自动调用removeObserver

这个不是与你博文提到的使用者不用显示调用反注册方法矛盾吗？
LiveData具有生命周期感知能力，所以LiveDataBus只需要调用注册回调方法，而不需要显示的调用反注册方法。

postValueDelay里面有错误

@Override
       public void postValueDelay(T value, long delay, TimeUnit unit) {
           mainHandler.postDelayed(new PostValueTask(value), unit.convert(delay, unit));
       }

应该改为

@Override
        public void postValueDelay(T value, long delay, TimeUnit unit) {
            mainHandler.postDelayed(new PostValueTask(value), TimeUnit.MILLISECONDS.convert(delay, unit));
        }

有没有key的追踪插件

为啥修改源码

这种修改源码的只能参考了

多线程会出现丢失消息的情况

多个线程频繁的发布消息，会出现丢失消息的情况

从非激活到激活状态收不到消息

引用的版本： implementation 'com.jeremyliao:live-event-bus:1.2.3'
MainActivity 代码

class MainActivity : AppCompatActivity() {

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        LiveEventBus.get().lifecycleObserverAlwaysActive(false)

        LiveEventBus.get().with("a").observe(this, Observer {
            Log.d("liveeventbus", "raw get")
        })
    }
}

SecondActivity 代码

class SecondActivity: AppCompatActivity() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        sendBtn.text = "发送通知"
        sendBtn.setOnClickListener {
            LiveEventBus.get().with("a").setValue("")
        }
    }
}

点击按钮发送后，关闭页面回到 MainActivity Log.d("liveeventbus", "raw get")方法并没有调用。
调试跟踪发现，回到 MainActivity 时 在 onChange 中被 isFilterChanged 过滤了。过滤的方法是 onStateChange

建议跨进程消息支持配置前台广播队列

对于跨进程通信既然是用广播实现的，有没有考虑到广播队列的效率问题？Android系统中默认使用的广播消息队列是后台广播消息，如果使用的应用非常多的话，不排除会出现阻塞问题，那么效率就是个问题了。另一方面，前台广播队列使用频率非常低，如果偶尔用一下对一些比较关键的消息的及时送达还是很有必要的。因此建议增加一个接口可以配置使用前台广播队列，要不然，我可能得自己改造一下了~~
@JeremyLiao

support sticky

hi: 支持sticky吗？

setValue方法丢失消息，发送5次相同的消息，只能收到3条，必现的

用的是live-event-bus版本，从fragment往Activity发消息

不适用androidX的组件 该库还有用嘛

针对第一版本的LiveDataBus的问题一些看法。

https://www.cnblogs.com/meituantech/p/9376449.html
如果当时结合ViewModel使用，应该可以避免接收到之前的消息。
`public class ViewModelUtil {

private static final ViewModelUtil instnce = new ViewModelUtil();

private ViewModelUtil() {}

public static ViewModelUtil getInstnce() {
    return instnce;
}

private TestViewModel testViewModel;

public TestViewModel getTestViewModel(FragmentActivity fragmentActivity) {
    if (testViewModel != null && testViewModel.clear) {
        testViewModel = null;
    }
    if (testViewModel == null) {
        testViewModel = ViewModelProviders.of(fragmentActivity).get(TestViewModel.class);
    }
    return testViewModel;
}

}`

`public class TestViewModel extends ViewModel {

public boolean clear;

private MutableLiveData<TestBean> liveData;

public TestViewModel() {
    liveData = new MutableLiveData<>();

}

public MutableLiveData<TestBean> getLiveData() {
    return  liveData;
}

@Override
protected void onCleared() {
    super.onCleared();
    clear = true;
    liveData = null;
}

}`

其实就是onCleared方法中把LiveData重新设为null

LiveDataBus中的bus这个map对象数据是否会一直持有，很多的时候是不是一种内存泄露？

延迟发送消息问题

大牛小哥哥
有个小问题
就是呢，当我在订阅之前发送一个延时消息，然后呢在这个延时消息发送之前订阅，SO就会收到这个消息
我看了一下代码，延时消息是通过Handler发出来的……

我是小菜鸟，见笑了……

这个反射涉及到混淆，最好提一下

设置对应的变量方法不混淆吧

viewholder oncreateview中加监听首次进入页面时无法收到回调，第二次才能正常收到回调

关于不接收订阅之前的消息的思考

你好：如果不接收之前的消息，我们让刚注册的观察者的onChanged方法不执行应该就可以了吧，用个标志位控制一下比如。

private class BusMutableLiveData<T> extends MutableLiveData<T>{

        /**
         * 是否需要更新数据,当主动调用setValue或者postValue的时候才触发
         */
        private boolean isChangeData = false;

        @Override
        public void setValue(T value) {
            isChangeData = true;
            super.setValue(value);
        }

        @Override
        public void postValue(T value) {
            isChangeData = true;
            super.postValue(value);
        }

        @Override
        public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<? super T> observer) {
            super.observe(owner, new ObserverWrapper<T>(observer,this));
        }
    }

    private class ObserverWrapper<T> implements Observer<T>{

        private Observer<? super T> mObserver;
        private BusMutableLiveData<T> mLiveData;

        public ObserverWrapper(Observer<? super T> observer,BusMutableLiveData<T> liveData) {
            mObserver = observer;
            mLiveData = liveData;
        }

        @Override
        public void onChanged(T t) {
            if(mLiveData.isChangeData&&mObserver!=null){
                mObserver.onChanged(t);
            }
        }
    }

Viewmodel

这个liveeventbus是不是不需要配合viewmodel是用啊，任何类型model都可以吧。

重复监听问题

两个ActivityA B 中都监听同一个key，用的是普通observer,A和B分别都会更新数据库，key是用来监听更新结果的。从A跳到B，在B中更新数据库后，再从B返回A的时候会触发A里面的监听.

Androidx 1.4.0版本

BusLiveData 不兼容 MediatorLiveData

        public void observeForever(@NonNull Observer<T> observer) {
            if (!observerMap.containsKey(observer)) {
                observerMap.put(observer, createForeverObserver(observer));
            }
            super.observeForever(observerMap.get(observer));
        }

现在这个过滤导致对消息的组合监听出现问题，调用链为

MediatorLiveData.onActive 或 MediatorLiveData.addSource
-> Source.plug
-> liveData.observeForever(此时的 liveData 为 busLiveData)
-> ObserverWrapper.onChange(带过滤功能的 wrapper )
-> 被过滤了...

如果上面的过滤规则一定要存在，则建议增加判断条件，如果调用链中有 MediatorLiveData 则不进行过滤。

请问为何要过滤掉由observeForever时产生的onChange

java.lang.NoSuchMethodException: get [class java.lang.Object]

Method methodGet = classObservers.getDeclaredMethod("get", Object.class);
[val methodGet = classObservers.getDeclaredMethod("get", Any::class.java)]
这个反射是否会出错？我这边试了一下，打包出来后会出现这个错误，不知道是不是和混淆有关。但看了dex文件，好像挺正常的。

没找到DEMO源码啊

没找到DEMO源码啊，能发一下DEMO源码吗？

类型转换存在隐患

您好，如果我提前注册一个接受Boolean的订阅者，然后在子线程发送一个String型消息，会直接报类型转换crash。
此类隐患建议避免

加油

加油，看好谷歌官方的【 Lifecycle +LiveData + ViewModel +Room】的架构组件

在同一个Acticity中接收俩个消息 Key无效

无论我使用postValue 或者是 setValue 在一个activity中向另一个activity中发送消息 Key值不同 ，但是在第二个activity 中的fragment俩个 observeSticky 方法均执行。

Cannot add the same observer with different lifecycles

不如增加一个支持 android.x的分支吧 我看到有人修改好了

在fragment里调用observe 是不能感应生命周期的吗

在fragment里调用observe 是不能感应生命周期的 只能在activity里能自动感应生命周期是吗，那这样说fragment里只能observeForever来手动取消了是么

关于观察者进入非活跃状态期间的消息全部抛弃的功能

处于某些原因我有一个自定义的观察者，在某些情况下我会将他设置为非活跃状态。
我希望在非活跃状态期间这个观察者不接收任何消息并且在恢复活跃的时候也不接收之前发送过的消息，只接收下一条新消息。
我在LiveEventBus-V2的基础上做了一些尝试，但是并没有达成目的，不知道可否指点一下思路。

请问该项目在有没有项目在应用了

混淆后不能用

混淆后，出现不能收到发送的信息，混淆用的readme提供的混淆

非 UI 线程订阅事件

非 UI 线程如果需要订阅事件，拿不到LifecycleOwner，怎么解？

黏性事件

你好，请问黏性事件发送出去，只要有一个接收就会移除掉吗？其他再注册就收不到了

gradle5以后，没有android.arch.lifecycle.Observer

我用的是Gradle5.1.1的。构建之后。没有android.arch.lifecycle.Observer，这个库就用不了。。。忘up主早日升级gradle5。我也多看看源码，看看能否直接引用源码，然后自己改改。

请问美团的modular-event没有开源吗？

modular-event

混淆了并不能用啊

-dontwarn com.jeremyliao.liveeventbus.**
-keep class com.jeremyliao.liveeventbus.LiveEventBus { *; }
-keep class android.arch.lifecycle.ExternalLiveData { *; }
-dontwarn android.arch.lifecycle.LiveData
-keep class android.arch.lifecycle.LiveData { *; }
-keep class android.arch.lifecycle.LifecycleRegistry { *; }
-keep class android.arch.core.internal.SafeIterableMap { *; }

似乎没有解决博客中说的 “订阅者会收到订阅之前发布的消息”

改了demo跑了一下，log打印出反射得到的version已经被更新了,是我的理解有错误咩..
改动：

    public void sendMsgByPostValue() {
        Log.i("wxd", "注册ob2 ");
        LiveDataBus.get()
                .with("key1", String.class)
                .observe(this, new Observer<String>() {
                    @Override
                    public void onChanged(@Nullable String s) {
                        Log.i("wxd", "ob2 onChanged: "+s);
                    }
                });
//        Observable.just(new Random())
//                .map(new Func1<Random, String>() {
//                    @Override
//                    public String call(Random random) {
//                        return "Message By PostValue: " + random.nextInt(100);
//                    }
//                })
//                .subscribeOn(Schedulers.io())
//                .subscribe(new Action1<String>() {
//                    @Override
//                    public void call(String s) {
//                        LiveDataBus.get().with("key1").postValue(s);
//                    }
//                });
    }

log:

11-28 17:50:34.090 13904-13904/com.jeremyliao.livedatabus I/wxd: ob1 onChanged: Message By SetValue: 71
11-28 17:50:35.961 13904-13904/com.jeremyliao.livedatabus I/wxd: ob1 onChanged: Message By SetValue: 43
11-28 17:50:36.372 13904-13904/com.jeremyliao.livedatabus I/wxd: ob1 onChanged: Message By SetValue: 50
11-28 17:50:37.087 13904-13904/com.jeremyliao.livedatabus I/wxd: ob1 onChanged: Message By SetValue: 93
11-28 17:50:40.279 13904-13904/com.jeremyliao.livedatabus I/wxd: 注册ob2 
11-28 17:50:40.280 13904-13904/com.jeremyliao.livedatabus I/wxd: ob2 onChanged: Message By SetValue: 93
11-28 17:50:40.280 13904-13904/com.jeremyliao.livedatabus I/wxd: hook:  lastV=3 objV=3

使用observe接口添加的Observer，用removeObserver无法正确移除

使用observe接口添加的Observer，用removeObserver无法正确移除，后续还是会收到事件。
使用observeForever就没问题。