[Rx,Kotlin] zip 연산자로 복수개(list)의 single 결합

zip 연산자를 통해 Observable을 동시에 실행하고 결과값을 받아 처리할 수 있다.

 

	val single1 = Single.just(1).delay(1000L, TimeUnit.MILLISECONDS)
	val single2 = Single.just(2).delay(1000L, TimeUnit.MILLISECONDS)
	val single3 = Single.just(3).delay(1000L, TimeUnit.MILLISECONDS)
	Single.zip(single1, single2, single3, Function3<Int, Int, Int, Int> { s1, s2, s3 ->
	s1 + s2 + s3
	}).subscribe({
	println("결과값 : $it")
	}, {
	it.printStackTrace()
	})
	Thread.sleep(2000L)
	println("Process finished")

view raw multipleSingle1.kt hosted with ❤ by GitHub

이때 FunctionN 은 ~9 까지 지원하는데

 

만약 Observable이 컴파일단에 정해지지 않고 동적으로 할당되거나 갯수가 10개가 넘는 경우에는 위와 같은 방법으로 어려울 수 있고

다음과 같은 방법으로 사용할 수 있다.

	val singles = (0..20).map {
	Single.just(it).delay(1000L, TimeUnit.MILLISECONDS)
	}.toList()
	Single.zip(singles, Function<Array<*>, Int> { array ->
	val result = array.filterIsInstance<Int>().reduce { acc: Int, i: Int -> acc + i }
	result
	}).subscribe({
	println("결과값 : $it")
	}, {
	it.printStackTrace()
	})
	Thread.sleep(2000L)
	println("Process finished")

view raw multipleSingle2.kt hosted with ❤ by GitHub

Observable(single) 자체로 구성된 리스트를 zip에 넣고, zip 내에서는 array로 받아 사용할 수 있다.

만약 Any타입으로 zip 내에서 사용되기 때문에 타입변환 등 타입에 대해 민감 할 경우 future를 사용하는 방법도 있다.

	val singles = (0..20).map {
	Single.just(it).delay(1000L, TimeUnit.MILLISECONDS)
	}.toList()
	Single.fromCallable {
	singles.map { it.toFuture() }
	}.flatMap {
	Single.just(
	it.map {
	it.get()
	}.reduce { acc, i -> acc + i }
	)
	}.subscribe({
	println("결과값 : $it")
	}, {
	it.printStackTrace()
	})
	Thread.sleep(2000L)
	println("Process finished")

view raw multipleSingle3.kt hosted with ❤ by GitHub

기존의 single 리스트를 toFuture시키면서 일괄 진행시키고, 합쳐야 될 상황에 get으로 비동기가 완료되고 반환된 값을 받아 처리할 수 있다.