RxJava有四个基本概念

• 被观察者 Observable

• 观察者 Observer
• 订阅 Subscribe
• 事件

事件回调方法

• onNext( ) , 相当于onClick( )

• onCompleted( )
• onError( )

基本实现

1.创建观察者 (Observer) -- 事件触发后的行为

Observer接口的实现方式:

Observer<String> observer = new Observer<String>() { 
@Override 
public void onNext(String s) {
 Log.d(tag, "Item: " + s); 
} 

@Override 
public void onCompleted() { 
Log.d(tag, "Completed!"); 
} 

@Override
 public void onError(Throwable e) { 
Log.d(tag, "Error!"); 
}

};

Subscriber 基本使用方式:

Subscriber<String> subscriber = new Subscriber<String>(){
@Override 
public void onNext(String s) { 
Log.d(tag, "Item: " + s); 
} 

@Override 
public 
void onCompleted() { 
Log.d(tag, "Completed!"); 
} 

@Override 
public void onError(Throwable e) { 
Log.d(tag, "Error!"); 
}

};

Observer 会被先转换成Subscriber再使用，两者基本使用方式一样。
区别:

1. onStart(): 这是 Subscriber增加的方法。
   它会在 subscribe 刚开始，而事件还未发送之前被调用，可以用于做一些准备工作，例如数据的清零或重置。这是一个可选方法，默认情况下它的实现为空。
   需要注意的是，如果对准备工作的线程有要求（例如弹出一个显示进度的对话框，这必须在主线程行， onStart()就不适用了，因为它总是在 subscribe 所发生的线程被调用，而不能指定线程。要在指定的线程来做准备工作，可以使用 doOnSubscribe()方法，具体可以在后面的文中看到。
2. unsubscribe(): 这是 Subscriber所实现的另一接口Subscription的方法，用于取消订阅。
   在这个方法被调用后，Subscriber将不再接收事件。一般在这个方法调用前，可以使用 isUnsubscribed()先判断一下状态。
   unsubscribe()这个方法很重要，因为在 subscribe()后， Observable会持有 Subscriber的引用，这个引用如果不能及时被释放，将有内存泄露的风险。
   所以最好保持一个原则：要在不再使用的时候尽快在合适的地方（例如 onPause() onStop() 等方法中）调用unsubscribe()来解除引用关系，以避免内存泄露的发生。

2.创建被观察者 (Observable)

决定什么时候触发，触发怎样的事件

Observable observable = Observable.create(new Observable.OnSubscribe<String>() { 
@Override 
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("Hello"); 
subscriber.onNext("Hi"); 
subscriber.onNext("Aloha"); 
subscriber.onCompleted(); }
});

OnSubscribe会被存储在返回的 Observable对象中，它的作用相当于一个计划表
当 Observable被订阅的时候 call()方法会自动被调用，事件序列就会依照设定依次触发
这样，由被观察者调用了观察者的回调方法，就实现了由被观察者向观察者的事件传递，即观察者模式。

快捷创建事件队列:

• just(T...): 将传入的参数依次发送出来。

Observable observable = Observable.just("Hello", "Hi", "World");
// 将会依次调用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();

• from(T[ ])/ from(Iterable<? extends T>) : 将传入的数组或 Iterable拆分成具体对象后，依次发送出来

String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);
// 将会依次调用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();

3.订阅(Subscribe)

observable.subscribe(observer);
或者：
observable.subscribe(subscriber);

或者不完整定义的回调, 自动创建Subscriber:

Action1<String> onNextAction = new Action1<String>() { 
// onNext() 
@Override 
public void call(String s) { 
Log.d(tag, s); 
}};

Action1<Throwable> onErrorAction = new Action1<Throwable>() { 
// onError() 
@Override 
public void call(Throwable throwable) { 
// Error handling 
}};

Action0 onCompletedAction = new Action0() { 
// onCompleted() 
@Override 
public void call() { 
Log.d(tag, "completed"); 
}};

// 自动创建 Subscriber ，并使用 onNextAction 来定义onNext()
observable.subscribe(onNextAction);

// 自动创建 Subscriber ，并使用 onNextAction 和 onErrorAction 来定义 onNext() 和 onError()
observable.subscribe(onNextAction, onErrorAction);

// 自动创建 Subscriber ，并使用 onNextAction、 onErrorAction 和 onCompletedAction 来定义 onNext()、 onError() 和 onCompleted()
observable.subscribe(onNextAction, onErrorAction, onCompletedAction);