비동기 처리를 쉽게 해주는 Coroutine에 대하여 알아보자_2(최종)

지난번 포스팅과 연결괴는 포스팅이므로 이전 글과 연결해서 봐주시면 감사하겠습니다.

(비동기 처리를 쉽게 해주는 Coroutine에 대하여 알아보자_1)

비동기 처리를 쉽게 해주는 Coroutine에 대하여 알아보자

이번 포스팅에서는 지난 코루틴 포스팅에 이어서 Channel에 대하여 알아보겠습니다.

 

Channel

Coroutine은 비동기처리를 쉽게 해주는 패턴이라고 이전 포스팅을 통해 배웠는데, 이전 포스팅에서는 한가지 코루틴에서만 데이터 처리가 이루어 지도록 하는 경우에 대하여 배웠습니다. 하지만 경우에 따라 Channel을 이용해 코루틴1의 데이터 값을 코루틴2로 넘겨 다수의 코루틴 수행이 가능하도록 만들어 줄 수 있는데요.

우선 Channel은 데이터와 코루틴이 1: 다수의 관계를 가질수 있게 만들어 준다라는 개념 알고 빠르게 에제를 통해 살펴 보겠습니다.

 

//Channel 기본 예제
        runBlocking {
            println("Start")
            val channel = Channel<Int>()
            launch {
                for (i in 1..5) {
                    channel.send(i + i)
                }
            }
                repeat(5){
                    val result = channel.receive()
                    println("$result")
                }

                println("Finish")

      	}
        
실행 결과 : 
Start
2
4
6
8
10
Finish

위의 예제를 통헤 데이터의 흐름을 지켜보자면

Int 타입의 channel을 하나 생성해주고 해당 채널에 for문을 통하여 1+1이라는 연산값을 하단의 repeat부분의 result변수에서 수신받습니다. 그 결과값을 출력하고 다시 위로 올라가 2+2를 똑같은 방식으로 전달 -> 다음은 3+3 ...이후 마지막으로 5+5까지의 코루틴 작업을 마치고 마지막 Finish를 출력하게됩니다.

코루틴은 작업을 빠르게 원하는 단계만큼 번갈아 수행하는 것임을 생각하면 repeat을 1번만 했을때는 가장 첫 작업결과만 출력된다는 것을 알수있으므로, Channel을 이용해 2개의 코루틴을 실행시킬경우 결과값 또한 For문이나 repeat등을 이용해 결과를 받아야합니다.

 

이처럼 Channel을 통하면 여러 코루틴을 한가지 데이터로 수행 시킬 수 있는데, Channel의 경우 데이터를 보내는 쪽과 받는쪽이 서로 더 처리할 작업이 남았는지 모르게 되는 경우가 생깁니다. 이러한 경우 .Close()메서드를 통해 더 이상 작업해야 할 내용이 없음을 알려줄수 있습니다. 그럼 예제를 통해 알아보겠습니다.

 

//Channel .Cancel() 예제
        runBlocking {
            println("Start")
            val channel = Channel<Int>()
            launch {
                for (i in 1..5) {
                    channel.send(i + i)
                }
                channel.close()
            }
            for(i in channel) println(i)

            println("Finish")

        }
        
실행 결과 : 
Start
2
4
6
8
10
Finish

 

위의 예제를 보면, 이전 예제와는 다르게 결과값을 받는 부분이 얼마나 반복해야 하는지 모르는 상태입니다. 하지만 For문안의channel.send()를 통해 데이터를 5번보내주었고 For문이 끝나자 .cancel()을 통해 더이상 보낼 데이터가 없다고 알려주었습니다.

이에 결과를 출력하는 result()쪽은 .cancel메서드가 호출되자마자 작업이 끝났음을 알고 코루틴을 나가 Finish라는 완료 문구를 띄우게 됩니다. 만약 위에 .cancel()메서드를 호출해주지 않았다면 2,4,6,8,10까지 연산을 마쳤다 하더라도 작업 종료 메서드인 .cacel()이 호출되지않았기 때문에 마지막 완료 문구 Finish를 띄우지 않고 영영 대기하게 됩니다.

 

Channeldp 대한 포스팅은 이쯤 마무리하고, 코루틴을 함수화 하여 사용하는 방법을 알아보겠습니다.

 

//코루틴 함수로 사용 예제
        runBlocking {
            println("Start")
            val squares = procedureSquares()
            squares.consumeEach {
                println("$it")
            }
            println("Finish")
        }
    }
    private fun CoroutineScope.procedureSquares() : ReceiveChannel<Int> = produce {
        for(x in 1..5) send(x+x)
    }
    
실행 결과 : 
Start
2
4
6
8
10
Finish

예제를 살펴보면 함수로 만들어준 procedureSquares()를 통해 결과값을 가져오는것을 알수 있습니다. 함수화 시킨다 라는것만 알면되기때문에 이부분은 설명을 딱히 하지 않겠습니다.

 

마지막으로 코루틴을 파이프라인으로 묶어 처리하는 방법에 대하여 알아 볼것인데요.

들어가기 앞서 Pipeline의 뜻을 알아보겠습니다.

나무위키를 통해 검색해보면,

파이프라인(영어: pipeline)은 한 데이터 처리 단계의 출력이 다음 단계의 입력으로 이어지는 형태로 연결된 구조.

라고 명시되어있습니다. 즉 쭉 이어져있는 파이프들에 물이 파이프를 통해 쭉 연결되어 흘러가듯이 코루틴을 통해 나온 데이터 결과 값을 다른 코루틴에 바로 입력해주고 이러한 행위를 원하는 만큼 반복 할 수 있게 여러 코루틴을 묶어서 사용하는 방법입니다.

 

그럼 어떻게 코루틴을 파이프라인으로 묶어서 사용하는지 알아보겠습니다.

 

우선 하나의 코루틴 작업의 결과 값을 다른 코루틴 하나에 입력하여 결과값을 얻어내는 예제를 알아보겠습니다.

 

 runBlocking {
         println("Start")

         val numbers = productnumbers()
         val squares = squares(numbers)

         for (i in 1..5){
             println("Result = ${squares.receive()}")
         }
         println("Finish")

         coroutineContext.cancelChildren()
     }

    }

    private fun CoroutineScope.productnumbers() = produce<Int> {
        var x = 1
        while (true){
            println("Send $x - productnumbers")
            send(++x)
            delay(100)
        }
    }

    private fun CoroutineScope.squares(numbers : ReceiveChannel<Int>) : ReceiveChannel<Int> = produce{
        for (x in numbers){
            println("Send $x - squares")
            send(x*x)
        }
    }
    
실행 결과 : 
Start
Send 1 - productnumbers
Send 2 - squares
Result = 4
Send 2 - productnumbers
Send 3 - squares
Result = 9
Send 3 - productnumbers
Send 4 - squares
Result = 16
Send 4 - productnumbers
Send 5 - squares
Result = 25
Send 5 - productnumbers
Send 6 - squares
Result = 36
Finish

위의 예제를 보면 코루틴의 확장함수 productnumbers, squares 2가지가 있고, 코루틴을 실행 시키는 runBlocking{ ... }안에 For문을 통해 1 부터 5까지의 정수값을 입력시켜주고 있습니다. 따라서 productnumbers에 1이 들어가고 ++을 통해 입력값을 1 증가 시킨뒤 send()를 통해 데이터를 발송시켜줍니다. 발송된 데이터는 receivechannel 확장함수인 squares에 입력되어지고 squares코루틴의 연산을 통해 나온 결과값을 다시 send(발송)하게 됩니다. 이렇게 발송된 결과 값 데이터는 runblock내부의 결과값 출력 부분인 result = squares.receive()에서 수신받게 되고 그 결과값을 출력하게 되는겁니다. 

위와 같은 연산을 1 부터 5까지 연산을 마친 후 더 이상 발송할 데이터가 없음을 나타내는 coroutineContext.childrenCancel()을 호출하여 coroutine을 종료하고 Finish를 출력하게 됩니다.

(흐름이 이해가 안간다면 ++x를 x++로 바꿔본다 던지 여러가지 형식으로 바꾸어서 실행하보면 도움이 됩니다.)

 

이처럼 코루틴 확장 함수 2가지를 통해 결과값을 받는 코루틴예제를 살펴보았는데 마지막으로 데이터생성하는 코루틴과 연산하는 코루틴을 같이 사용하는 방법까지만 살펴보고 마치겠습니다.

        //확장함수를 이용한 코루틴 함수 파이프라인 구축 사용 예제2(데이터 생성 , 다중 코루틴)
        runBlocking {
            val producer = product()
            repeat(5){
                launch(it, producer)
            }
            delay(1000L)
            producer.cancel()
        }
    }

    fun CoroutineScope.launch(id : Int, channel: ReceiveChannel<Int>){
        launch {
            for (msg in channel){
                println("launch $id received $msg ")
            }
        }
    }
    //확장함수를 이용한 코루틴 함수 파이프라인 구축 사용 예제2(데이터 생성 , 다중 코루틴)
    private fun CoroutineScope.product() = produce<Int> {
        var x = 1
        while (true){
            println("Send $x - product")
            send(x++)
            delay(100)
        }
    }
    
실행 결과 : 
Send 1 - product
launch 0 received 1 
Send 2 - product
launch 0 received 2 
Send 3 - product
launch 1 received 3 
Send 4 - product
launch 2 received 4 
Send 5 - product
launch 3 received 5 
Send 6 - product
launch 4 received 6 
Send 7 - product
launch 0 received 7 
Send 8 - product
launch 1 received 8 
Send 9 - product
launch 2 received 9 
Send 10 - product
launch 3 received 10

위의 예제를 보면 runblock{ ... }안에서 product라는 데이터를 데이터 생성 코루틴 products를 통해 생성해 주었습니다.

생성 이후 send를 통해 receivechannel 확장 함수인launch에 데이터를 전달하고 데이터 값을 1 증가 시켜주었고 이러한 작업을 1000L밀리초까지 반복하여 결과값을 얻어내는 예제입니다.

하지만 데이터 처리 시간의 차이 때문인지 repeat의 it값이 처음에 0이 두번나오는데 이 부분을 필자는 repeat이 한번 돌때마다 작업이 완료 될때까지 기다리도록 joinAll()메서드를 사용하여 수정하였고 그 결과 순차적으로 repeat값이 0,1,2,3,4와 같이 잘 나오는것을 확인 할 수 있었습니다. 이처럼 비동기식의 처리의 경우 여러가지 작업을 한번에 수행하기 떄문에 혹 같은 데이터를 여러 코루틴을 이용해 작업한다면 데이터 처리 시간에 특히 주의해서 사용해야 안전합니다.

 

마치며...

이렇게 두번의 포스팅으로 나누어 Coroutine에 대하여 알아보았는데요. 비동기식 프로그래밍은 오래전부터 중요시 여겨지는 프로그래밍 패턴중 한가지입니다. 그도 그럴것이 동기식 프로그램의 경우 한번에 한가지일을 끝날때까지 하는것이고 비동기식은 한번에 여러가지 일을 한번에 하는것이기 때문에 데이터 처리에 용이하며, 특히 코루틴의 경우 runBlocking{ ... }등과 같은 메서드를 통해 내가 원하는 시간 / 조건 동안만 작업을 할 수 있게 만들어 줄 수 있을 뿐만 아니라, 요즘 많이 올라오는 TestCode를 작성할때 아주 편리하게 사용됩니다. 하지만 잘못된 순서로 연산이 처리될경우 원하는 결과 값이 안나올 가능성도 있기 떄문에 잘 알아두고 적절하게 쓸수 있도록 평소에 연습을 해두면 좋을거 같습니다. 주구장창 예제와 설명만 써놓은 코루틴 포스팅은 이쯤에서 마무리 하도록 하겠습니다!

 

 

*공부를 하며 작성한 포스팅이므로 틀린 내용이 있거나 부족할 수 있습니다. 부족한 부분에 대하여 또는 잘못된 정보에 대한 지적은 감사하겠습니다!*