Warum Intervall Operator in RxJava die Ergebnisse von flatMap ändern

Question

ich diesen Code leite:

private void concatenatedSets() { 
    Observable<String> concatenatedSets = 
      Observable.just("1/5/8", "1/9/11/58/16/", "9/15/56/49/21"); 

    concatenatedSets.flatMap(s -> Observable.from(s.split("/"))) 
      .map(s -> Integer.valueOf(s)) 
      .interval(200, TimeUnit.MILLISECONDS) 
      .subscribeOn(Schedulers.computation()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(new Subscriber<Long>() { 
       @Override 
       public void onCompleted() { 

       } 

       @Override 
       public void onError(Throwable e) { 

       } 

       @Override 
       public void onNext(Long aLong) { 
        tvCounter.setText(String.valueOf(aLong)); 
       } 
      }); 
} 

Die erwartete Ausgabe

1 
5 
8 
1 
9 
11 
58 
16 
9 
15 
56 
49 
21 

Jedoch wäre, alles, was ich bekommen ist eine endlose Sitzung des Zählens von 1 bis unendlich, wie in 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12 ...

glaube ich das Problem kam, nachdem ich

01.235.164 hinzugefügt

.interval(200, TimeUnit.MILLISECONDS) 

aber ich weiß nicht warum.

Answer 1

interval() ist eine Factory-Methode auf Observable, die Ihnen eine Sequenz von Werten gibt, beginnend mit 0, in regelmäßigen Abständen. Es ist ein Quellenoperator, der statisch wie Observable.interval() aufgerufen werden sollte, aber Sie haben es für eine Instanz aufgerufen. Die meisten IDEs sollten Sie warnen, eine statische Methode für eine Instanz aufzurufen, da normalerweise etwas anderes passiert als erwartet.

Ich vermute, Sie wollen Werte getrennt von 200 ms aufspalten; In RxJava gibt es dafür keinen Standardoperator, aber hier ist eine allgemeine Frage zu SO (und sollte schon eine Antwort haben).

Answer 2

Normalerweise verwende ich einen Trick, der die Liste der Elemente kombiniert, um mit Intervall mit zip-Operator zu emittieren. Schau dir den Code an.

@Test 
public void concatenatedSets() { 
    Subscription subscription = 
      Observable.just("1/5/8", "1/9/11/58/16/", "9/15/56/49/21") 
        .flatMap(s -> Observable.zip(Observable.from(s.split("/")), Observable.interval(200, TimeUnit.MILLISECONDS), (i, t) -> i)) 
        .map(Integer::valueOf) 
      .subscribeOn(Schedulers.computation()) 
      .observeOn(Schedulers.io()) 
        .subscribe(System.out::println); 
    new TestSubscriber((Observer) subscription) 
      .awaitTerminalEvent(1000, TimeUnit.MILLISECONDS); 
} 

Dies wird jeder Artikel 200ms

emittiert verzögern Sie können praktischeres Beispiel sehen hier https://github.com/politrons/reactive