RxJava 2 Erste Schritte: Aufgaben seriell ausführen. `andThen` oder` defer`?

Question

Ich möchte zwei Schritte in RxJava seriell ausführen. Ich möchte ein Schritt vor Schritt 2 beginnt wie zu beenden:

step 1: start 
step 1: finish 
step 2: start 
step 2: finish 

Ich versuche Variationen verschiedene API und RxJava meine zwei Schritte parallel ausgeführt wird, die nicht das Verhalten, das ich will:

step 1: start 
step 2: start 
step 2: finish 
step 1: finish 

Im folgenden Codebeispiel versuche ich beide andThen und defer und ich bekomme parallele Ausführung. Wie kann ich das beheben, so dass ein Schritt nach dem erfolgreichen Abschluss des anderen ausgeführt wird?

Der Methodenname andThen impliziert sequenzielle serielle Ausführung. Die Methode defer nimmt eine Funktion, die eine andere Completable erzeugt, die die Methodensignatur ist, die ich für die Ausführung der seriellen Task, die ich haben möchte, erwarte. Gib mir auch nicht das Ergebnis, das ich will.

Muss ich auf Observable/Flowable umstellen? Oder kann ich zwei Schritte mit Completable verketten?

public class RxStep1Then2 { 
    public static Completable simulateCompletable(ScheduledExecutorService es, String msg, int msDelay) { 
     System.out.println(String.format("%s: start", msg)); 

     ScheduledFuture<?> future = es.schedule(() -> { 
      System.out.println(String.format("%s: finish", msg)); 
     }, msDelay, TimeUnit.MILLISECONDS); 

     return Completable.fromFuture(future); 
    } 

    public static void rxMain(ScheduledExecutorService es) { 
//  Completable c = simulateCompletable(es, "step 1", 1000) 
//    .andThen(simulateCompletable(es, "step 2", 500)); 

     Completable c = simulateCompletable(es, "step 1", 1000) 
       .defer(() -> simulateCompletable(es, "step 2", 500)); 

     c.blockingAwait(); 
     System.out.println("blockingAwait done"); 
    } 

    public static void main(String[] args) throws Exception { 
     ScheduledExecutorService es = Executors.newScheduledThreadPool(5); 
     System.out.println("Started ExecutorService."); 

     rxMain(es); 

     es.shutdown(); 
     es.awaitTermination(5, TimeUnit.MINUTES); 
     System.out.println("Shutdown ExecutorService. Done."); 
    } 
} 

Answer 1

Die parallele Ausführung geschieht, weil Ihre simulateCompletable die Aufgaben beginnt, bevor die Completable s selbst erstellt werden. Sie könnten eine verzögerte Completable direkt verwenden:

Completable.fromAction(() -> System.out.println("First")) 
    .delay(1, TimeUnit.SECONDS) 
    .andThen(Completable.fromAction(() -> System.out.println("Second"))) 
    .blockingAwait(); 

Beachten Sie, dass

Completable c = simulateCompletable(es, "step 1", 1000) 
      .defer(() -> simulateCompletable(es, "step 2", 500)); 

nicht Kette jede Operation, weil defer eine statische Factory-Methode, die eine unabhängige Completable schafft; der erste simulateCompletable 's Completable ist einfach verloren.