EventBus与RxJava

2022-04-06 09:22:32

总的来说，EventBus是一款针对Android优化的发布/订阅事件总线，主要功能是替代Intent,Handler,BroadCast在Fragment，Activity，Service，线程之间传递消息。而Rxjava则是一种基于异步数据流的处理方案。如果一个订阅者需要注册多个事件的时候，Rxjava需要一个个单独的注册，而EventBus则可以实现一个订阅者订阅多个事件，和一个事件对应多个订阅者。

EventBus

EventBus是一个Android端优化的publish/subscribe消息总线，简化了应用程序内各组件间、组件与后台线程间的通信。比如请求网络，等网络返回时通过Handler或Broadcast通知UI，两个Fragment之间需要通过Listener通信，这些需求都可以通过EventBus实现。EventBus仅仅适合当做组件间的通讯工具使用，主要用来传递消息，避免搞出一大堆的interface。

使用

添加依赖
使用EventBus之前需要添加相关的依赖：

compile 'org.greenrobot:eventbus:3.0.0'

然后在onStart()方法中注册它，在onStop()方法中消耗。

    //注册eventBus
    @Override
    protected void onStart() {
        super.onStart();
        EventBus.getDefault().register(this);
    }

    //取消注册
    @Override
    protected void onStop() {
        super.onStop();
        EventBus.getDefault().unregister(this);
    }

在需要接受的地方添加订阅者（使用@Subscribe注解），@Subscribe注解来描述一个public无返回值的非静态方法，注解后面可以跟threadMode，来给定订阅者处理事件所在的线程。

@Subscribe(threadMode = ThreadMode.MAIN)
    public void onEventMainThread(IdEvent event) {
        if (event != null) {
            
        }
    }

EventBus包含4个ThreadMode：

ThreadMode.POSTING
事件的处理在和事件的发送在相同的进程，所以事件处理时间不应太长，不然影响事件的发送线程，而这个线程可能是UI线程；
ThreadMode.MAIN
事件的处理会在UI线程中执行,事件处理不应太长时间；
ThreadMode.BACKGROUND
事件的处理会在一个后台线程中执行，尽管是在后台线程中运行，事件处理时间不应太长；
ThreadMode.ASYNC
事件处理会在单独的线程中执行，主要用于在后台线程中执行耗时操作，每个事件会开启一个线程（有线程池），但最好限制线程的数目。

例如：

/**
     * 在后台线程中执行，如果当前线程是子线程，则会在当前线程执行，如果当前线程是主线程，则会创建一个新的子线程来执行
     * @param event
     */
    @Subscribe(threadMode = ThreadMode.BACKGROUND)
    public void  onEventBackgroundThread(MessageEvent event){
        System.out.println("onEventBackgroundThread::"+" "+Thread.currentThread().getName());
    }

    /**
     * 创建一个异步线程来执行
     * @param event
     */
    @Subscribe(threadMode = ThreadMode.ASYNC)
    public void onEventAsync(MessageEvent event){
        System.out.println("onEventAsync::"+" "+Thread.currentThread().getName());
    }

    /**
     * 在主线程中运行
     * @param event
     */
    @Subscribe(threadMode = ThreadMode.MAIN)
    public void onEventMain(MessageEvent event){
        System.out.println("onEventMain::"+" "+Thread.currentThread().getName());
    }

    /**
     *默认的线程模式，在当前线程下运行。如果当前线程是子线程则在子线程中，当前线程是主线程，则在主线程中执行。
     * @param event
     */
    @Subscribe(threadMode = ThreadMode.POSTING)
    public void onEventPosting(MessageEvent event){
        System.out.println("onEventPosting::"+" "+Thread.currentThread().getName());
    }

发布者是在主线程还是子线程，发布的消息在所有定义好的实体类型订阅者中都可以接收到消息。也就是，可以实现一个订阅者订阅多个事件，和一个事件对应多个订阅者。

EventBus.getDefault().post(new MessageEvent("hello"));

RxJava

RxJava 在 GitHub 主页上的自我介绍是 "a library for composing asynchronous and event-based programs using observable sequences for the Java VM"，翻译成中文是，一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库，也就是一个异步事件库。
RxJava 的优势即是简洁，但它的简洁的与众不同之处在于，随着程序逻辑变得越来越复杂，它依然能够保持简洁。

使用

使用RxJava之前需要先添加相关的依赖：

compile 'io.reactivex.rxjava2:rxjava:2.1.8'
compile 'io.reactivex.rxjava2:rxandroid:2.1.8'

使用RxJava之前，有以下几个概念需要注意：

Observeable(被观察者)/Observer（观察者）
Flowable(被观察者)/Subscriber(观察者)

//被观察者在主线程中，每1ms发送一个事件
Observable.interval(1, TimeUnit.MILLISECONDS)
                //.subscribeOn(Schedulers.newThread())
                //将观察者的工作放在新线程环境中
                .observeOn(Schedulers.newThread())
                //观察者处理每1000ms才处理一个事件
                .subscribe(new Action1() {
                      @Override
                      public void call(Long aLong) {
                          try {
                              Thread.sleep(1000);
                          } catch (InterruptedException e) {
                              e.printStackTrace();
                          }
                      }
                  });

2.x中提供了以上两种观察者与被观察者的关系。
Observeable用于订阅Observer，是不支持背压的，而Flowable用于订阅Subscriber，是支持背压(Backpressure)的。
背压的概念是：指在异步场景中，被观察者发送事件速度远快于观察者的处理速度的情况下，一种告诉上游的被观察者降低发送速度的策略。

Flowable<String> t = Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> e) throws Exception {
                e.onNext("hello Rx2.0.");
                e.onComplete();
            }
        }, BackpressureStrategy.BUFFER);

一般而言，上游的被观察者会响应下游观察者的数据请求，下游调用request(n)来告诉上游发送多少个数据。这样避免了大量数据堆积在调用链上，使内存一直处于较低水平。

我们需要调用request去请求资源，参数就是要请求的数量，一般如果不限制请求数量，可以写成Long.MAX_VALUE。如果你不调用request，Subscriber的onNext和onComplete方法将不会被调用。

Subscriber subscriber = new Subscriber() {
            @Override
            public void onSubscribe(Subscription s) {
                System.out.println("onSubscribe()");
                s.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(Object o) {
                System.out.println(""+o.toString());
            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onComplete() {

            }
        };

除了上面这两种观察者，还有一类观察者：

Single/SingleObserver：返回泛型数据的结果给观察者
Completable/CompletableObserver：返回完成的结果
Maybe/MaybeObserver ：前两者的复合体

Rxjava内置 Scheduler

Schedulers.immediate() ：默认的，直接在当前线程运行；
Schedulers.newThread() ：启用新线程，在新线程工作；
Schedulers.io()：I/O操作（读写文件，读写数据库，网络信息交互等）、和newThread（）最大的区别是：io()内部实现是一个无数量上限的线程池，可以重用空闲的线程；
Schedulers.computation()：计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在
computation() 中，否则 I/O 操作的等待时间会浪费 CPU。
AndroidSchedulers.mainThread()： Android专用的，指定的操作在Android的主线程运行。

subscribeOn()和observerOn()

subscribeOn() 指定subscribe() 所发生的线程，即 Observable.OnSubcribe被激活时所处的线程，或者叫事件产生的线程。
observerOn() 指定Subscriber 运行所在的线程，或者叫做事件消费的线程。

码农公寓

EventBus

使用

RxJava

使用

Rxjava内置 Scheduler

subscribeOn()和observerOn()

相关文章