如何在RxJava上使用線程？

Question

 Observable.create(new Observable.OnSubscribe<Integer>() { 

     public void call(final Subscriber<? super Integer> obs) { 
      new Thread(){ 
       public void run(){ 
        obs.onNext(1); 
       } 
      }.start(); 
      new Thread(){ 
       public void run(){ 
        obs.onNext(2); 
       } 
      }.start(); 
      new Thread(){ 
       public void run(){ 
        obs.onNext(3); 
       } 
      }.start(); 
      obs.onCompleted(); 
     } 
    }).subscribe(new Subscriber<Integer>(){ 

     public void onCompleted() { 
      System.out.println("Complete"); 

     } 

     public void onError(Throwable arg0) { 
      // TODO Auto-generated method stub 

     } 

     public void onNext(Integer arg0) { 
      System.out.println(arg0); 
      try { 
       Thread.sleep(1000); 
      } catch (InterruptedException e) { 
       // TODO Auto-generated catch block 
       e.printStackTrace(); 
      } 
     } 

    }); 

正如你看到的，我想要做的多線程上的Java與RxJava

我已經在谷歌研究了很多資源，但大部分都是針對Android

任何人能告訴什麼是最簡單的方法來實現它？

我覺得我的代碼是一個小：<

Answer 1

嗯，這是一個有點不清楚你在問什麼，但它是值得的，至少有超過可觀察的類草草讀，因爲它有很多的方法，將使你的生活更輕鬆。例如，你的代碼大致翻譯爲：

Observable 
.just(1,2,3) 
.subscribeOn(Schedulers.io()) 
.zipWith(Observable.interval(1000,1000,TimeUnit.milliseconds), 
    (item, pos) -> item) 
.doOnCompleted(() -> System.out.println("Complete"); 
.subscribe(item -> System.out.println(item)); 

但是，你首先需要放在你想要做什麼明確的 - 你可以把它的話呢？