RxJava - flatmap vs concatMap - neden abonelikte sipariş veriyorsunuz?

Question

this thread conCatMap ve düzmap'ye göre, yalnızca öğelerin gönderildiği sıraya göre farklılık gösterir. Bu yüzden bir test yaptım ve basit bir tamsayı akışı yarattım ve hangi sırada yayılacağını görmeyi istedim. Ben 1-5 arasında bir sayı ve iki tarafından iki katına çıkarılacak küçük bir gözlemlenebilir yaptım. Kolay.

myObservable.flatMap(new Func1<Integer, Observable<Integer>>() { 
     @Override 
     public Observable<Integer> call(Integer integer) { 
      return Observable.just(integer * 2); 
     } 
    }).subscribe(new Observer<Integer>() { 
     @Override 
     public void onCompleted() { 

     } 

     @Override 
     public void onError(Throwable e) { 

     } 

     @Override 
     public void onNext(Integer integer) { 
     Log.v("myapp","from flatMap:"+integer); 
     } 
    }); 

ve concatMap kullanarak tam aynı kod: Burada

flatmap ile kodudur i günlüklerinde çıktısını gördü sipariş için aynı idi

myObservable.concatMap(new Func1<Integer, Observable<Integer>>() { 
     @Override 
     public Observable<Integer> call(Integer integer) { 
      return Observable.just(integer * 2); 
     } 
    }).subscribe(new Observer<Integer>() { 
     @Override 
     public void onCompleted() { 

     } 

     @Override 
     public void onError(Throwable e) { 

     } 

     @Override 
     public void onNext(Integer integer) { 
     Log.v("myapp","from concatmap:"+integer); 
     } 
    }); 

, neden ? Sadece concatMap'in düzeni koruyacağını düşündüm.

Answer 1

Gördüğünüz şey bir tesadüf. Her zaman flatMap bir değer döndürür, bir öncekiyle aynı iş parçacığı üzerinde yapar. Ben çoklu yararlanmak için örnek değiştirdiniz

:

Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10) 
     .flatMap(integer -> Observable.just(integer) 
       .observeOn(Schedulers.computation()) 
       .flatMap(i -> { 
        try { 
         Thread.sleep(new Random().nextInt(1000)); 
         return Observable.just(2 * i); 
        } catch (InterruptedException e) { 
         e.printStackTrace(); 
         return Observable.error(e); 
        } 
       })) 
     .subscribe(System.out::println, 
       Throwable::printStackTrace, 
       () -> System.out.println("onCompleted")); 

Farklı düzeni zorlamak için rastgele bir gecikme ile her 2 * i değerini geciktirmek ediyorum. Ayrıca, bundan önce observeOn(Schedulers.computation()) ekledim, böylece sonraki operatör (flatMap) hesaplama iş parçacığı havuzu üzerinde çalışır - bu çoklu iş parçacığı büyüsünü yapar. Ben concatMap ile just sonra flatMap değiştirin sonra bir düzgün sıralı çıktı elde

I/System.out: 6 
I/System.out: 4 
I/System.out: 12 
I/System.out: 14 
I/System.out: 8 
I/System.out: 2 
I/System.out: 16 
I/System.out: 20 
I/System.out: 10 
I/System.out: 18 
I/System.out: onCompleted 

:

Bu

I (Android'de) benim örneğin olsun çıkıştır. Uygun bir açıklama ile great post by Thomas Nield var.