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

//观察者模式,这里产生事件,事件产生后发送给接受者,但是一定要记得将事件的产生者和接收者捆绑在一起,否则会出现错误
Observable.create(new ObservableOnSubscribe<String>() {
  @Override
  public void subscribe(ObservableEmitter<String> e) throws Exception {
    //这里调用的方法会在产生事件之后会发送给接收者,接收者对应方法会收到
    e.onNext("hahaha");
    e.onError(new Exception("wulala"));
    e.onComplete();
  }/*--
}).subscribe(new Observer<String>() {
  //接受者,根据事件产生者产生的事件调用不同方法
  @Override
  public void onSubscribe(Disposable d) {
    Log.e(TAG, "onSubscribe: ");
  }

  @Override
  public void onNext(String value) {
    Log.e(TAG, "onNext: " + value);
  }

  @Override
  public void onError(Throwable e) {
    Log.e(TAG, "onError: ", e);
  }

  @Override
  public void onComplete() {
    Log.e(TAG, "onComplete: ");
  }
});

     subscribe();     //表示发送者随意发送数据,接受者什么都不管,什么都不接收.
     subscribe(Consumer<? super T> onNext) {}   //只响应onNext()事件,其他的事件忽略.
     subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {}     //含义同上
     subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}     //含义同上
     subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}   //含义同上

Observable.create(new ObservableOnSubscribe<String>() {
  @Override
  public void subscribe(ObservableEmitter<String> e) throws Exception {
    e.onNext("hahaha");
    e.onNext("hahaha");
    e.onNext("hahaha");
    Log.e(TAG,"运行在什么线程" + Thread.currentThread().getName());
    e.onComplete();
  }
}).subscribeOn(Schedulers.newThread())        //线程调度器,将发送者运行在子线程
 .observeOn(AndroidSchedulers.mainThread())     //接受者运行在主线程
 .subscribe(new Observer<String>() {
  @Override
  public void onSubscribe(Disposable d) {
    Log.e(TAG, "onSubscribe: ");
    Log.e(TAG,"接收在什么线程" + Thread.currentThread().getName());
  }

  @Override
  public void onNext(String value) {
    Log.e(TAG, "onNext: " + value);
  }

  @Override
  public void onError(Throwable e) {
    Log.e(TAG, "onError: ", e);
  }

  @Override
  public void onComplete() {
    Log.e(TAG, "onComplete: ");
  }
});