Ինչպե՞ս եք կառավարում backpressure խնդիրը RxJava-ում։

RxJava- ում backpressure- ը կառավարելու համար օգտագործվում են օպերատորներ, որոնք նախատեսված են Flowable և Observable տիպերի համար, որոնք կարող են արտադրել մեծ քանակությամբ տարրեր:

• Backpressure ռազմավարություններ (Flowable-ի համար): Նշում են, թե ինչպես մշակել ավելցուկային տարրերը: Հիմնական ռազմավարությունները՝
  • MISSING: Հատուկ ռազմավարություն չի կիրառվում, ենթադրվում է, որ սպառողը ինքն է կարգավորում:
  • ERROR: Եթե տարրերը գալիս են ավելի արագ, քան սպառողը կարող է մշակել, գեներացվում է MissingBackpressureException:
  • BUFFER: Բուֆերավորում է բոլոր ավելցուկային տարրերը մինչև դրանք պահանջվեն: Կարող է հանգեցնել հիշողության սպառման:
  • DROP: Դատարկում է այն տարրերը, որոնք չեն կարող անմիջապես մշակվել:
  • LATEST: Պահպանում է միայն վերջին ավելցուկային տարրն, նախորդները մերժում:
• Օպերատորներ, որոնք աջակցում են backpressure (Flowable-ի համար): Օգտագործվում են հոսքերը փոխակերպելու համար, որոնք չեն աջակցում backpressure (օրինակ՝ Observable), կամ հատուկ ռազմավարությունների իրականացման համար: Օրինակներ՝
  • toFlowable(): փոխարկում է Observable-ը Flowable-ի՝ նշված backpressure ռազմավարությամբ:
  • onBackpressureBuffer(): իրականացնում բուֆերային ռազմավարություն:
  • onBackpressureDrop(): իրականացնում մերժման ռազմավարություն:
  • onBackpressureLatest(): իրականացնում վերջին տարրին պահելու ռազմավարություն:
• Հոսքերը կառավարող օպերատորներ (Flowable և Observable): Կարող են սահմանափակել տարրերի քանակը կամ մշակման արագությունը: Օրինակներ՝
  • throttleFirst(): Ուղարկում է առաջին տարր խմբից, որը գալիս է նշված միջակայքում:
  • throttleLast() / debounce(): Ուղարկում է վերջին տարրն ընդմիջումից հետո:
  • sample(): Կանոնավոր կերպով վերցնում է վերջին ստացված տարր:
  • limit(): Սահմանափակում է արտանետվող տարրերի քանակը:
  • buffer(): Ամբողջացնում է տարրերը նշված չափի կամ ժամանակային միջակայքի բուֆերներում:

Ընտրությունը կոնկրետ մոտեցման կախված է տվյալների աղբյուրի բնույթից, մշակման պահանջներից և թույլատրելի տվյալների կորուստից:

// Օրինակ `onBackpressureDrop` օգտագործելու
Flowable.range(1, 100000)
    .onBackpressureDrop() // Դատարկում է ավելցուկային տարրերը
    .observeOn(Schedulers.computation())
    .subscribe(
        item -> {
            // Ապակաս արագ մշակման իմիտացիա
            try {
                Thread.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("Մշակված՝ " + item);
        },
        error -> System.err.println("Սխալ՝ " + error.getMessage()),
        () -> System.out.println("Ավարտվեց")
    );

// Observable- ի փոխարկում Flowable-ի` BUFFER ռազմավարությամբ
Observable.range(1, 100000)
    .toFlowable(BackpressureStrategy.BUFFER)
    .observeOn(Schedulers.io())
    .subscribe(
        item -> {
            // Մշակում
            System.out.println("Տարրերի մշակումը՝ " + item);
        },
        error -> System.err.println("Սխալ՝ " + error.getMessage())
    );