RxJava

Question

ile sınır verimi Şu anda içine girdiğim durumun açıklanması oldukça zor, bu yüzden sadece sorunu açıklamak için daha basit bir versiyon yazacağım.

Dosyaların bir ArrayList tarafından tanımlanan bir dosya sırası yayan bir Observable.from()10 var. Bu dosyaların tümü bir sunucuya yüklenmelidir. Bunun için işi yapan bir işleve sahibim ve bir Observable döndürür.

Observable<Response> uploadFile(File file); 

Saçma alır bu kodu çalıştırmak

, Observable.from() tüm dosyaları yayar ve onlar olanları hiç yüklenen ya da en azından işleyebilir parçacığı bir maks edilir.

Paralel olarak en fazla 2 dosya yüklemesi almak istiyorum. Bunu benim için halledebilecek bir operatör var mı?

Ben tampon, pencere ve bazı diğerleri çalıştı ama onlar sadece yerine sürekli iki paralel dosya yüklemeleri sahip iki maddeyi birbirine yayarlar görünüyor. Ayrıca, yükleme bölümünde bir maksimum iş parçacığı havuzu ayarlamayı denedim, ancak bu durum benim durumumda kullanılamaz.

Bu hak için basit bir operatör olmalıydı? Bir şey mi eksik?

Answer 1

Tüm dosyalar aynı anda tüm dönüşümleri yürüten flatMap() kullandığınız için paralel olarak karşıya yüklendiğini düşünüyorum. Bunun yerine, bir dönüşümü diğerinden sonra çalışan concatMap() kullanmalısınız. Ve iki paralel yüklemeyi çalıştırmak için, görüntülenen dosyalara window(2) numaralı telefonu çağırmanız ve flatMap() kodunu kodunuzda yaptığınız gibi çağırmanız gerekir.

Observable<Response> responses = 
    files 
     .window(2) 
     .concatMap(windowFiles -> 
     windowFiles.flatMap(file -> uploadFile(file)); 
    ); 

GÜNCELLEME:

Ne istediğini tam olarak ne yapar ki, daha iyi bir çözüm buldum. Eşzamanlı iş parçacığının maksimum sayısını kabul eden flatMap() bir aşırı yükleme var.

Observable<Response> responses = 
    files 
     .onBackpressureBuffer() 
     .flatMap(index -> { 
     return uploadFile(file).subscribeOn(Schedulers.io()); 
     }, 2);