RxJava Observable 필터링

HEETAE HEO·2022년 7월 31일

이번 글에서는 필터 연산자들을 이용해 원하는 값들 만을 걸러내는 기능들을 구현해보겠습니다.

filter()

filter()의 경우 발행되는 데이터 스트림에서 특정 연산식의 조건에 부합하는 아이템들만 발행해주는 연산자입니다.

즉 조건의 부합되지않는 데이터는 제거해주고 원하는 데이터만 정의한 동작을 수행하도록 해줍니다.

fun main() {
	Observable.just(1,2,3,4,5,6,7,8)
    .filter { it % 2 == 0 }
    .subscribe(System.out::println)
}

다음과 같이 짝수들만 출력이 되는 것을 볼 수 있습니다.

take()

take(i)의 경우 지정한 개수(i만큼)만큼 데이터를 발행해줍니다.

fun main(){
	Observable.just(1,2,3,4,5,6,7,8)
    .take(3)
    .subscribe(System.out::println)

1
2
3

skip()

skip(i)의 경우에는 첫번째 부터 i번째 데이터를 생략하고 Observable에게 보내줍니다. 즉 take()와는 반대되는 동작을 수행합니다.

fun main(){
	Observable.just(1,2,3,4,5,6,7,8)
    .skip(3)
    .subscribe(System.out::println)
}

distinct()

distinct()의 경우 데이터의 중복을 제거해줍니다.

fun main(){
	Observable.just(1,2,3,4,1,2,3,5)
    .distinct()
    .subscribe(System.out::println)

sample()

sample()는 일정 시간 간격으로, 최근에 Observable이 발행한 아이템들을 발행합니다.

fun main() {
    Observable.interval(100, TimeUnit.MILLISECONDS)
        .sample(300, TimeUnit.MILLISECONDS)
        .subscribe(System.out::println)
}

100ms 마다 데이터를 발행하는 interval(100, TimeUnit.MILLISECONDS) 연산자에, 300ms 마다 데이터를 샘플링하는 연산자 sample(300, TimeUnit.MILLISECONDS) 를 붙여 다음과 같은 결과가 나오게 됩니다.

takeLast()

take()와는 반대 개념으로 뒤에서 부터 몇 개의 데이터를 발행할지를 지정해주는 메서드입니다.

fun main() {
    Observable.just(1, 2, 3, 4, 5, 6, 7, 8)
        .takeLast(3)
        .subscribe(System.out::println)
}

6
7
8

debounce()

debounce()의 경우 특정 시간동안 더 이상 데이터가 발행되지 않을 때에 데이터을 발행하는 연산자입니다.

가장 많이 사용되는 곳이 검색하는 부분에서 입니다.

검색창 입력필드에 사용자가 검색어를 입력할 때 검색버튼을 누르지 않아도 검색되는 기능을 구현한다는 가정하에 나이키 라는 검색어를 타이핑할 때 ㄴ -> 나 -> 나ㅇ -> 나이 다음과 같이 검색어의 변경에 따라 검색을 요청하는 값을 받아오는 동작을 수행해야합니다. 그렇기에 debounce() 이용해 특정 시간 동안 검색어에 변경이 없다면 다음과 같이 변경하는 것 입니다.

fun main() {
    Observable.create { emitter: ObservableEmitter<Any?> ->
        emitter.onNext("1")
        Thread.sleep(100)
        emitter.onNext("2")
        emitter.onNext("3")
        emitter.onNext("4")
        Thread.sleep(100)
        emitter.onNext("5")
        emitter.onNext("6")
        emitter.onNext("7")
        emitter.onNext("8")
        emitter.onNext("9")
        emitter.onNext("10")
        Thread.sleep(100)
    }
        .debounce(10, TimeUnit.MILLISECONDS)
        .subscribe {
            println(it)
        }
}