[리엑티브 연습] concat, zip (네트워크 통신 예제)

안녕하세요. 블랙진입니다.

리엑티브 연습 포스팅입니다.

conact

여러 옵저버블의 데이터를 이터러밍 없이 발행해 줍니다. 즉 순서를 보장해 주는 오퍼레이터 입니다.

언제 사용할까요?

여러 API를 통신했을 때 순차적으로 처리가 필요한 경우에 사용합니다. 한 예로 사용자 정보를 가져오고 가져온 정보를 사용해 두번 째 api를 통신해야 되는 상황에서 사용하면 좋습니다.(다시 생각해보니 이 상황에서는 concatMap이 더 어울릴 것 같네요 ㅎㅎ)

예제

val temp1 = true
val temp2 = "BlackJin"
val temp3 = arrayOf(1,2,3,4,5)

val api1 = Single.just(temp1).delay(300, TimeUnit.MILLISECONDS) // 300ms 후 boolean을 반환하는 통신
val api2 = Single.just(temp2).delay(100, TimeUnit.MILLISECONDS) // 100ms 후 String 을 반환하는 통신
val api3 = Single.just(temp3).delay(600, TimeUnit.MILLISECONDS) // 600ms 후 배열을 반환하는 통신

val start = System.currentTimeMillis() // 통신 시간 측정을 위한 변수

3개의 통신 예제가 있습니다. 각각의 딜레이는 300ms, 100ms, 600ms를 가지고 있죠? 이 예제 API를 가지고 테스트를 진행해 보겠습니다.

1. api1, api2 통신

Single.concat(api1, api2)
    .subscribe({
        println("${System.currentTimeMillis() - start} : $it")
    }){
        println(it.message)
    }

concat 연산자는 통신의 순서를  보장해 줍니다.

[결과]

388 : true

488 : BlackJin

1. boolean를 반환하는 api1은 300ms가 걸리므로 388ms 에 true를 반환합니다.

2. String을 반환하는 api2는 100ms가 걸리는데 api1이 동작되고 나서 진행되기 때문에 488ms(388 + 100) 에 BlackJin을 반환했습니다.

2. api1, api2, api3 통신

Single.concat(api1, api2, api3)
    .subscribe({
        println("${System.currentTimeMillis() - start} : $it")
    }){
        println(it.message)
    }

이번에는 600ms가 걸리는 api3 추가해 보겠습니다.

[결과]

395 : true

496 : BlackJin

1100 : [Ljava.lang.Integer;@690bc184

1. boolean를 반환하는 api1은 300ms가 걸리므로 395ms 에 true를 반환합니다.

2. String을 반환하는 api2는 100ms가 걸리는데 api1이 동작되고 나서 진행되기 때문에 496ms(395 + 101) 지난 후에 BlackJin을 반환했습니다.

3. Array를 반환하는 api3는 600ms가 걸리는데 api1, api2가 모두 동작되고 진행되기 때문에 1100ms(496 + 604) 지난 후에 Array를 반환했습니다.

zip

여러 옵저버블의 결과를 합쳐서 하나의 데이터를 발행합니다. 이때 기준이 되는 시간은 가장 긴 옵저버블입니다.

언제 사용할까요?

여러 API를 동시에 통신하여 하나의 결과로 합쳐줄 때 사용하면 됩니다. 만일 zip을 사용하지 않으면 통신이 끝나는 시점이 모두 다르고 일부 통신은 연결중에 실패 할 수도 있습니다. 그러면 멀티 스레딩 환경에서 데이터의 동기화가 잘 이뤄지지 않고 어떤 사이드 이펙트가 발생할지 알 수 없는 상황이 됩니다.

예제

1. api1, api2 통신

Single.zip(
    api1,
    api2,
    BiFunction<Boolean, String, String>
    { t1, t2 -> "$t1 $t2" }
)
    .subscribe({
        println("${System.currentTimeMillis() - start} : $it") 
    }) {
        println(it.message)
    }

[결과]

323 : true BlackJin

api1(delay: 300ms)과 api2(delay: 100ms) 중에서 api1의 통신이 끝난 시점(323ms)에 데이터가 발행 됨을 확인할 수 있습니다. 

1. api1, api2, api3 통신

Single.zip(
    api1,
    api2,
    api3,
    Function3<Boolean, String, Array<Int>, String>
    { t1, t2, t3 -> "$t1 $t2 $t3" }
)
    .subscribe({
        println("${System.currentTimeMillis() - start} : $it")
    }) {
        println(it.message)
    }

[결과]

626 : true BlackJin [Ljava.lang.Integer;@51c02284

api1(delay: 300ms), api2(delay: 100ms), api3(delay: 600) 중에서 api3의 통신이 끝난 시점(626ms)에 데이터가 발행 됨을 확인할 수 있습니다. 

@Test
fun test() {

    val temp1 = true
    val temp2 = "BlackJin"
    val temp3 = arrayOf(1,2,3,4,5)

    val api1 = Single.just(temp1).delay(300, TimeUnit.MILLISECONDS) // boolean 을 반환하는 통신
    val api2 = Single.just(temp2).delay(100, TimeUnit.MILLISECONDS) // String 을 반환하는 통신
    val api3 = Single.just(temp3).delay(600, TimeUnit.MILLISECONDS) // 배열을 반환하는 통신

    val start = System.currentTimeMillis() // 통신 시간 측정을 위한 변수

    Single.zip(
        api1,
        api2,
        BiFunction<Boolean, String, String>
        { t1, t2 -> "$t1 $t2" }
    )
        .subscribe({
            println("${System.currentTimeMillis() - start} : $it") // 321ms
        }) {
            println(it.message)
        }

    Single.zip(
        api1,
        api2,
        api3,
        Function3<Boolean, String, Array<Int>, String>
        { t1, t2, t3 -> "$t1 $t2 $t3" }
    )
        .subscribe({
            println("${System.currentTimeMillis() - start} : $it") //624ms
        }) {
            println(it.message)
        }

    Single.concat(api1, api2)
        .subscribe({
            println("${System.currentTimeMillis() - start} : $it")
            //389ms : true
            //490 : BlackJin
        }){
            println(it.message)
        }

    Single.concat(api1, api2, api3)
        .subscribe({
            println("${System.currentTimeMillis() - start} : $it")
            //389ms : true
            //490 : BlackJin
            //1102 : array
        }){
            println(it.message)
        }

    Thread.sleep(1300)
}