RxJava

Question

के साथ थ्रू थ्रूपुट जिस मामले में मैं अभी हूं, उसे समझा जाना काफी मुश्किल है, इसलिए मैं इस मुद्दे को समझाने के लिए एक सरल संस्करण लिखूंगा।

मेरे पास Observable.from() है जो फ़ाइलों की एक अनुक्रम को ArrayList फ़ाइलों द्वारा परिभाषित करता है। इन सभी फ़ाइलों को सर्वर पर अपलोड किया जाना चाहिए। इसके लिए मेरे पास एक ऐसा कार्य है जो नौकरी करता है और Observable देता है।

Observable<Response> uploadFile(File file); 

मुझे इस कोड की यह पागल हो जाता है चलाने के लिए, Observable.from() फ़ाइलों के सभी का उत्सर्जन करता है और वे लोगों को बिल्कुल भी अपलोड किया जाता है, या कम से कम धागे इसे संभाल कर सकते हैं की एक अधिकतम के लिए।

मैं समानांतर में अधिकतम 2 फ़ाइल अपलोड करना चाहता हूं। क्या कोई ऑपरेटर है जो मेरे लिए इसे संभाल सकता है?

मैं बफर, खिड़की और कुछ अन्य की कोशिश की लेकिन वे केवल बजाय दो समानांतर फ़ाइल अपलोड लगातार होने के एक साथ दो आइटम फेंकना करने लगता है। मैंने अपलोडिंग भाग पर अधिकतम थ्रेड पूल सेट करने का भी प्रयास किया, लेकिन इसका उपयोग मेरे मामले में नहीं किया जा सकता है।

इस अधिकार के लिए एक साधारण ऑपरेटर होना चाहिए? क्या मैं कुछ भूल रहा हूँ?

Answer 1

मुझे लगता है कि सभी फाइलें समानांतर में अपलोड की गई हैं क्योंकि आप flatMap() का उपयोग कर रहे हैं, जो एक साथ सभी परिवर्तनों को निष्पादित करता है। इसके बजाय आपको concatMap() का उपयोग करना चाहिए, जो एक के बाद एक रूपांतरण चलाता है। और दो समांतर अपलोड चलाने के लिए आपको window(2) पर कॉल करने की आवश्यकता है, और फिर आपके कोड में flatMap() को आमंत्रित करें।

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

अद्यतन:

मैं एक बेहतर समाधान पाया है, जो करता है आप क्या चाहते हैं। flatMap() का अधिभार है जो समवर्ती धागे की अधिकतम संख्या स्वीकार करता है।

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