在RxJava中调用onNext之间暂停

Question

如何使onNext方法调用之间的延迟？我有一个代码：

Observable<Integer> obs = Observable.range(1, 1000) 
        .subscribeOn(Schedulers.io()) 
        .observeOn(AndroidSchedulers.mainThread()); 
      subs = obs.subscribe(new Subscriber<Integer>() { 
       @Override 
       public void onNext(Integer number) { 
        mCurrentNumber = number; 
        mCounter.setText(String.valueOf(number)); 
       } 

       @Override 
       public void onCompleted() { 
        isRunning = false; 
       } 

       @Override 
       public void onError(Throwable e) { 

       } 
      }); 

例如，我需要设置一个暂停100毫秒

做到这一点是使用压缩到您观察到的结合与时间间隔观察到的延迟输出

Answer 1

那么你可以使用interval运营商在指定的时间间隔后，将发出的商品，需要操作人员使用的流项目，才需要数

Observable.interval(100, TimeUnit.MILLISECONDS) 
     .take(1000) 
     .subscribe(new Action1<Long>() { 
      @Override public void call(Long timerValue) { 

     } 
     }); 

Answer 2

的一种方式。

Observable.zip(Observable.range(1, 5) 
    .groupBy(n -> n % 5) 
    .flatMap(g -> g.toList()), 
Observable.interval(50, TimeUnit.MILLISECONDS), 
(obs, timer) -> obs) 
.doOnNext(item -> { 
    System.out.println(System.currentTimeMillis() - timeNow); 
    System.out.println(item); 
    System.out.println(" "); 
}).toList().toBlocking().first(); 

Answer 3

如果你想1,2,3...1000这个怎么样：

subs = Observable.interval(0, 100, TimeUnit.MILLISECONDS, Schedulers.io()) 
     .take(1000) 
     .map(counter -> counter++) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .subscribe(new Subscriber<Long>() { 
      @Override 
      public void onNext(Long number) { 
       mCurrentNumber = number; 
       mCounter.setText(String.valueOf(number)); 
      } 

      @Override 
      public void onCompleted() { 
       isRunning = false; 
      } 

      @Override 
      public void onError(Throwable e) { 

      } 
     }); 

• .interval将在IO
运行
• 这将立即开始发射物品0L, 1L, 2L......infinity。
• 的.take(1000)会第一时间1000个项目，然后停止（完成）
• 的.map(counter -> counter++)将改变以上1L, 2L,......infinity因为你从1在你的榜样请求1000，而不是0到1000

如果你想100,200,300然后使用.map(counter -> counter*100)

如果你想“样品”每100毫秒的范围内，从而创造一个位的随机数，然后使用：

Observable.range(1, 1000) 
     .subscribeOn(Schedulers.io()) 
     .sample(100, TimeUnit.MILLISECONDS) 
     ...... 
     //rest here