如何從一組線程中提取最快響應線程的結果？

Question

問題：考慮一個場景，其中有多個服務可以執行特定的任務。每項服務都可以用不同的時間響應任務。我們需要始終從速度最快的服務中選擇響應。

Answer 1

下面是使用RxJava Observables的示例工作代碼，該代碼打印來自一組線程的最快響應線程的結果。

public static void main(String[] args) { 
      // Create a slow thread which spans 5 secs 
      Callable<String> task1 = new Callable<String>() { 

        @Override 
        public String call() throws Exception { 
          Thread.sleep(5000); 
          return "task1"; 
        } 
      }; 
      // Create a faster thread which spans 1 secs 
      Callable<String> task2 = new Callable<String>() { 

        @Override 
        public String call() throws Exception { 
          Thread.sleep(1000); 
          return "task2"; 
        } 
      }; 
      List<Callable<String>> tasks = new ArrayList<>(); 
      tasks.add(task1); 
      tasks.add(task2); 

      String result = null; 
      try { 
        result = Observable.from(tasks) 
            .subscribeOn(Schedulers.computation()) 
            .flatMap(eachTask -> Observable.fromCallable(eachTask) 
                .subscribeOn(Schedulers.io()) 
                .doOnNext(e -> System.out.println("Executing your action on "+Thread.currentThread().getName())) 
                .doOnError(e -> System.out.println("Failed reason for : "+Thread.currentThread().getName()+" with error "+e.getMessage())) 
                ) 
            .toBlocking() 
            .first(); 

      } catch (Exception e) { 
        System.out.println(e.getMessage()); 
      } 
      System.out.println("result--->"+result); 
} 

Answer 2

如果我理解你的權利，你需要的東西是這樣的：

taskSource 
    .flatMap(task -> // for each task 
     Observable.merge(
      // submit same task to multiple services 
      service1.submit(task), 
      service2.submit(task), 
      ..., 
      serviceN.submit(task) 
      ) 
      .take(1)) // take first response; discard others 
    ... // continue processing result of the task 
    .subscribe(...) 

Answer 3

您希望Observable.amb操作。它具有與不止一種排放的Observables一起工作的好處。