如何在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)); 

但是，你首先需要放在你想要做什么明确的 - 你可以把它的话呢？