RxJava学习笔记（入门）

1.rxjava介绍

RxJava is a Java VM implementation of Reactive Extensions: a library for composing asynchronous and event-based programs by using observable sequences.

It extends the observer pattern to support sequences of data/events and adds operators that allow you to compose sequences together declaratively while abstracting away concerns about things like low-level threading, synchronization, thread-safety and concurrent data structures.

rxjava是用java实现Reactive Extensions，可观测的序列来组成异步的、基于事件的程序的库。它扩展了观察者模式来支持数据/事件序列，而抽象掉的事情，如低级别的线程，同步，线程安全和并发数据结构。（简而言之，就是做异步处理,类似于Android中的Asynctask和handler）

2.rxjava 简单使用

2.1 添加依赖

compile"io.reactivex.rxjava2:rxjava:2.0.7"

compile"io.reactivex.rxjava2:rxandroid:2.0.1"

2.2创建被观察者Observable

//被观察者 就是产生事件

Observable observable=Observable.create(new ObservableOnSubscribe() {

@Override

    public void subscribe(ObservableEmitter emitter)throws Exception {

//ObservableEmitter事件发射器

        emitter.onNext(1);

        emitter.onNext(2);

        emitter.onNext(3);

        emitter.onNext(4);//发射事件

    }

});

2.3创建观察者Observer

//观察者 就是处理事件

Observer observer=new Observer() {

     @Override

    public void onSubscribe(Disposable d) {

//这里开始采用subscribe 链接

        Log.e("onSubscribe",""+d.toString());

    }

  @Override

    public void onNext(Integer integer) {

//处理onnext事件

        Log.e("onNext",""+integer);

    }

    @Override

    public void onError(Throwable e) {

//处理错误的事件

        Log.e("onError",""+e.getMessage());

    }

    @Override

    public void onComplete() {

//完成事做出响应

        Log.e("onComplete","");

    }

};

2.4订阅Subscribe

//订阅

observable.subscribe(observer);

2.5使用Disposable切断链接

//观察者

Observer observer=new Observer() {

Disposabledisposable;

    @Override

    public void onSubscribe(Disposable d) {

//这里开始采用subscribe 链接

        Log.e("onSubscribe",""+d.toString());

        disposable=d;

    }

@Override

    public void onNext(Integer integer) {

//处理onnext事件

        Log.e("onNext",""+integer);

    }

@Override

    public void onError(Throwable e) {

//处理错误的事件

        Log.e("onError",""+e.getMessage());

        disposable.dispose();

    }

@Override

    public void onComplete() {

//完成事做出响应

        Log.e("onComplete","");

    }

};

3.切换线程

observable.subscribeOn(Schedulers.newThread())//指定被观察者线程 即 发送事件的线程

        .observeOn(AndroidSchedulers.mainThread())//指定观察者线程 即 处理事件时使用的线程

        .subscribe(observer);

如果observable.subscribeOn() 多次指定线程只有第一次的有效

observeOn每次切换都有效

4.其它操作符 

看这里->Android RxJava：这是一份全面 & 详细 的RxJava操作符 使用攻略