[RxJava2] 리액티브 RxJava2Extensions 라이브러리에 있는 연산자 : count(), max(), min(), sumInt(), averageDouble(), delay(), timeInterval()

공감 및 댓글은 포스팅 하는데  아주아주 큰 힘이 됩니다!! 포스팅 내용이 찾아주신 분들께  도움이 되길 바라며 더 깔끔하고 좋은 포스팅을  만들어 나가겠습니다^^

이번 포스팅에서는 RxJava2Extenions 라이브러리에 있는 연산자들 중

count(), max(), min(), sumInt(), averageDouble() 에 대해 공부합니다.

그리고 delay(), timeInterval() 에 대해서도 공부합니다.

count()

Observable에서 발행한 데이터의 갯수를 발행합니다.

결과 값은 항상 한 개의 값만 가지므로 Single<Long>으로 발행합니다.

max(), min()

이름처럼 최댓값과 최솟값을 발행합니다.

sumInt(), averageDouble()

이 함수들도 이름 그대로 입니다. 정수 값으로 합을 구한다.

실수 값으로 평균을 구한다.

위 5개의 함수를 모두 다뤄보는 예제입니다.

package rx.java.chapter04.etc;
 
import hu.akarnokd.rxjava2.math.MathFlowable;
import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.Single;
 
public class RxMathEx {
 
    public static void main(String[] args) {
        
        Integer[] numbers = {1, 2, 3, 4, 5};
        
        // count() 함수
        Single<Long> src1 = Observable.fromArray(numbers).count();
        src1.subscribe(System.out::println);
        
        
        // min(), max() 함수
        Flowable.fromArray(numbers)
        .to(MathFlowable::min)
        .subscribe(System.out::println);
        Flowable.fromArray(numbers)
        .to(MathFlowable::max)
        .subscribe(System.out::println);
        
        
        // sum(), average() 
        Flowable<Integer> src2 = Flowable.fromArray(numbers)
                .to(MathFlowable::sumInt);
        src2.subscribe(System.out::println);
 
        Flowable<Double> src3 = Flowable.fromArray(numbers)
                .to(MathFlowable::averageDouble);
        src3.subscribe(System.out::println);
        
    }
 
}
 

delay()

시간을 인자로 전달 받으며 유틸리티 연산자로서 보조적인 역할을 합니다.

package rx.java.chapter04.etc;
 
import java.util.concurrent.TimeUnit;
 
import io.reactivex.Observable;
 
public class DelayEx {
 
    public static void main(String[] args) {
 
        String[] strs = {"1", "2", "3", "4", "5"};
        
        // 데이터 발행 전 딜레이.
        Observable<String> src = Observable.fromArray(strs)
                .delay(100L, TimeUnit.MILLISECONDS);
        src.subscribe(System.out::println);
        try {
            Thread.sleep(1000);
        } catch (Exception e) {
            // TODO: handle exception
        }
        
    }
 
}
 

예제를 실행시켜보시면 알겠지만, 입력받은 시간만큼 데이터의 발행을

지연시켜주는 역할을 합니다.

timeInterval()

어떤 값을 발행했을 때 이전 값을 발행한 이후 시간이 얼마나 지났는지

알려주는 함수입니다.

package rx.java.chapter04.etc;
 
 
import io.reactivex.Observable;
import io.reactivex.schedulers.Timed;
 
public class TimeIntervalEx {
 
    public static void main(String[] args) {
 
        try {
 
        String[] strs = {"1", "3", "5", "7", "9"};
        
        Observable<Timed<String>> src = Observable.fromArray(strs)
                .delay(item -> {
                    Thread.sleep(100);
                    return Observable.just(item);    
                })
                .timeInterval();
        
        src.subscribe(System.out::println);
            Thread.sleep(1000);
        } catch (Exception e) {
            // TODO: handle exception
        }
    }
 
}
 

이상입니다. 다음 포스팅에서는 스케줄러에 대해 공부합니다.

감사합니다.