Rxjava简介

目录

1. Rx简介
2. Rxjava基础知识
3. Rxjava创建操作符
4. Rxjava的线程操作
5. Rxjava变换过滤操作符
6. Rxjava合并、链接操作符
7. Rxjava条件和布尔操作符

Rxjava由来

Rxjava是Rective Extensions的Java实现，用于通过使用Observable/Flowable序列来构建异步和基于事件的程序的库。

什么是Rx

Rx是Reactive Extensions的缩写，最初是LINQ的一个扩展，由微软开发。2012年开源，Rx是一个编程模型，目标是
提供一致的编程接口，帮助开发者方便的处理异步数据流。

Reactive Extensons的历史

.........

Rx模式

1:使用观察者模式：
创建：Rx可以方便的创建事件流和数据流。
组合：Rx可以查询式的操作符组合和变换数据流
监听：Rx可以订阅任何可以观察的数据流并执行操作。
2:简化代码
！！！Rxjava和lambde结合使用可以极大简化代码。

为什么选择Rxjava

1：Rx扩展了观察者模式:用于支持数据和事件序列，添加了一些操作符，可以使用声明式的方式来组合这些序列，无需关系底层实现。
如线程安全，同步，线程，并发数据结构和非阻塞I/O。
2：响应式编程：Rx提供了一系列的操作符，我们可以使用他们来过滤(filter),选择(select),变换(transform)，结合(combine)和组合(compose)多个Observable，这些操作让执行和复合变的非常高效，

Rx的observable：

Observable是一个抽象的概念，是异步双向的push，适用于任何场景，拥有他的近亲Iterable的全部优雅和灵活。Rx的Observable给传统的观察者模式添加了两种缺少的语义，这两种语义都需要我们主动去调用。

1. onError():将执行的错误发布给观察者
2. OnComplete():将执行完成的事件发布给观察者

Rxjava能做什么

Rxjava可以使用在任何地方，包括APP所依赖的底层框架，Android客户端，服务端开发等。

eg:某APP从服务器获得酒店列表，将价格大于500的房间显示到UI。

    //Observable的使用
    Observable.fromIterable(getHotelsFromServer())
        .flatMap(new Function<Hotel, Observable<Room>(){
            @Override
            public Observable<Room> apply(Hotel hotel){
                return Observable.fromIterable(hotel.rooms);
            }
        })
        .filter(new Predicate<Room>(){
            @Override
            public boolean test(@NonNull Room room) throws Exception{
                return room.price >= 500;
            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.MainThread())
        .subscribe(new Consumer<Room>(){
            @Override
            public void accept(Room room){
                displayUI(room);
            }
        });

先不用理解，简单看一下代码，后面的博客中会详细解释。

如果使用了lambda就会代码更少

     Observable.fromIterable(getHotelsFromServer())
        .flatMap(hotel -> Observable.fromIterable(hotel.rooms))
        .filter(room -> room.price >= 500)
        subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(this::displayUI);

Rxjava HelloWorld:

    Observable.create(new ObservableOnSubscribe<String>(){
        @Override
        public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception{
            e.onNext("hello world");
        }
    }).subscribe(new Consumer<String>(){
        @Override
        public void accept(@NonNull String e) throws Exception{
            System.out.println(e);
        }
    });
    
    //也可以这样
    Observable.just("hello world").subscribe(new Consumer<String> (){
        @Override
        public void accept(@NonNull String s) throws Exception {
            System.out.println(s);
        }
    })

后注：

目前在移动端可以大量看到Rxjava代码，也产生了许多框架，RxBinding、Retrofit、RxLifeCycle等。