使用rxJava超时实用程序时未获取TimeoutException

Question

如果5秒钟内未输入任何文本，则不会收到TimeoutException。下面的代码方法将调用getMsg（）并等待文本输入。我添加了“ timeout（5，TimeUnit.SECONDS）”，仅等待5秒钟进行输入。如果用户在5秒钟内未输入msg，我想显示超时错误。

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.concurrent.TimeUnit;

import rx.Observable;

public class TestRx {

   public static void main( String[] args ) throws IOException {
      Observable.just( getMsg() )
            .timeout( 5, TimeUnit.SECONDS )
            .subscribe( System.out::println,
                  e -> e.printStackTrace() );
      System.out.println( "End..." );
   }

   private static String getMsg() throws IOException {
      BufferedReader reader = new BufferedReader( new InputStreamReader( System.in ) );
      System.out.print( "Enter a msg:" );
      String msg = reader.readLine();
      return msg;
   }
}

Answer 1

getMsg()在您进入RxJava之前先执行。just()不会神奇地使括号中的代码以延迟的方式发生。您需要fromCallable：

public static void main( String[] args ) {
    Observable.fromCallable(() -> getMsg() )
        .timeout( 5, TimeUnit.SECONDS )
        .subscribe( System.out::println,
              e -> e.printStackTrace() );
    System.out.println( "End..." );
}

更新资料

阻塞发生在主线程上，该线程在此设置中不会中断。替代方法是使用subscribeOn并可能blockingSubscribe等待数据或终止：

Observable.fromCallable( () -> getMsg() )
          .subscribeOn(Schedulers.io())             // <----------------------
          .timeout( 5, TimeUnit.SECONDS )
          .blockingSubscribe( System.out::println,
                    e -> e.printStackTrace() );

System.out.println( "End..." );