三、创建操作符

• just（）讲一个或多个对象转换成发射这个或这些对象的一个0bservable
  Observable.just(1, 2, 3)
  .subscribe(new Subscriber<Integer>() {
  @Override
  public void onNext(Integer item) {
  Log.d(TAG,"Next: " + item);
  }

        @Override
        public void onError(Throwable error) {
            System.err.println("Error: " + error.getMessage());
        }
  
        @Override
        public void onCompleted() {
             Log.d(TAG,"Sequence complete.");
        }
    });
  

• from（）将一个Iteration，一个Future，或者一个数组转换成一个0bservable
  Javadoc: from(array)
  Javadoc: from(Iterable)
  Javadoc: from(Future)
  Javadoc: from(Future,Scheduler)
  Javadoc: from(Future,timeout, timeUnit)

  Integer[] items = { 0, 1, 2, 3, 4, 5 };
  Observable myObservable = Observable.from(items);
  myObservable.subscribe(
    new Action1<Integer>() {
        @Override
        public void call(Integer item) {
            Log.d(TAG, item+"");
        }
    },
    new Action1<Throwable>() {
        @Override
        public void call(Throwable error) {
            Log.d(TAG,"Error encountered: " + error.getMessage());
        }
    },
    new Action0() {
        @Override
        public void call() {
            Log.d(TAG,"Sequence complete");
        }
    }
   );
  

• repeat() 创建一个重复发射指定数据或数据序列的0bservable

• repeatWhen（）创建一个重复发射指定数据或数据序列的0bservable，他依赖于另一个0bservable发射的数据

• create（observer）您可以使用Create 操作符从头创建一个Observable 。您通过此操作符接受观察者作为其参数的函数。通过调用观察者的-因此，它表现为一个可观察写这个功能onNext， onError和onCompleted适当的方法。

  
  create.png

  //订阅者
   Subscriber subscriber= new Subscriber<Integer>() {
     @Override
     public void onNext(Integer item) {
         Log.d(TAG, "Next: " + item);
     }
     @Override
     public void onError(Throwable error) {
         Log.d(TAG, "Error: " + error.getMessage());
     }
     @Override
     public void onCompleted() {
         Log.d(TAG, "Sequence complete.");
     }
     };
  
      //create方法默认不在任何特定的调度器上执行。
     Observable observable = Observable.create(new      Observable.OnSubscribe<Integer>() {
     //当Observable.subscribe被调用时（有订阅者时）执行call方法
       @Override
       public void call(Subscriber<? super Integer> observer) {
          try {
          //检查观察者的isUnsubscribed状态，以便在没有观察者的时候，让Observable停止发射数据或者做昂贵的运算
         //      List<String> sList=      Arrays.asList(ConstantValues.ALLOW_CHANNEL_DOWN);
        //        Iterator<String> iterator= sList.iterator();
        //        try {
        //            while (iterator.hasNext()){
        //                if (observer.isUnsubscribed())return ;
        //                observer.onNext(iterator.next());
        //            }
        //            if (!observer.isUnsubscribed())
        //                observer.onCompleted();
        //        } catch (Exception e) {
        //            if (!observer.isUnsubscribed())observer.onError(e);
        //        }
          for (int i = 1; i < 5; i++) {
              if(i == 4){
                  //取消订阅 (Unsubscribing),调用这个方法表示你不关心当前订阅的Observable了，
                  //因此Observable可以选择停止发射新的数据项（如果没有其它观察者订阅）。
                  subscriber.unsubscribe();
              }
              if (!observer.isUnsubscribed()) {
                  observer.onNext(i);
              }
          }
          if (!observer.isUnsubscribed()) {
              observer.onCompleted();
          }
        } catch (Exception e) {
            observer.onError(e);
        }
      }
    } );
      //订阅
      observable.subscribe(subscriber);
  

-defer（）只有当订阅者订阅才创建0bservable；为每个订阅创建一个新的0bservable

• range（）创建一个按指定范围的整数序列的0bservable；

• interval（）创建一个按照给定时间间隔发射整数序列的0bservable
  private void op_Interval(TextView textView){
  //以秒为单位，每隔1秒发射一个数据
  Observable.interval(1, TimeUnit.SECONDS)
  //interva operates by default on the computation Scheduler,so observe on main Thread
  //如果需要更新view，要在主线程中订阅
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(new Subscriber<Long>() {
  @Override
  public void onCompleted() {
  Log.d(TAG,"onCompleted" );
  }
  @Override
  public void onError(Throwable e) {
  Log.d(TAG,"onError:" + e.getMessage());
  }
  @Override
  public void onNext(Long aLong) {
  Log.d(TAG,"interval:" + aLong);
  textView.setText("Interval:"+aLong);
  }
  });
  }

• timer（）创建一个在给定的演示之后发射单个数据的0bservable

   private void op_Timer(TextView textView){
    SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
    String startTime = sdf.format(new Date());
    Log.v(TAG, "startTime:" + startTime);
    Observable.timer(2, TimeUnit.SECONDS)
            .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Subscriber<Long>() {
        @Override
        public void onNext(Long item) {
            //Timer创建的对象在2秒钟后发射了一个0
            Log.d(TAG,"Next: " + item);
            String endTime =  sdf.format(new Date());
             textView.setText(startTime+":Timer："+endTime);
            Log.v(TAG, "endTime:" + endTime);
        }
        @Override
        public void onError(Throwable error) {
            System.err.println("Error: " + error.getMessage());
        }
        @Override
        public void onCompleted() {
            Log.d(TAG,"Sequence complete.");
        }
    });
  

• empty（）创建一个什么都不做直接通知完成的0bservable

• error（）创建一个什么都不做直接通知错误的0bservable

• never（）创建一个不发射任何数据的Observable

参考：http://blog.csdn.net/xmxkf/article/details/51645348