Wie extrahiere ich das Ergebnis aus dem am schnellsten antwortenden Thread aus einer Gruppe von Threads?

Question

Problem: Betrachten Sie ein Szenario, in dem mehrere Dienste eine bestimmte Aufgabe ausführen können. Jeder Dienst kann mit unterschiedlichem Zeitaufwand auf die Aufgabe reagieren. Wir müssen immer die Antwort von dem Dienst mit der schnellsten Leistung auswählen.

Answer 1

Unten ist der Beispielarbeitscode, der RxJava Observables verwendet, der das Ergebnis aus dem am schnellsten reagierenden Thread aus einer Gruppe von Threads druckt.

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

Wenn ich Sie recht verstanden, müssen Sie etwas wie folgt aus:

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

Sie wollen den Observable.amb Operator. Es hat den Vorteil, mit Observablen mit mehr als einer Emission zu arbeiten.