비동기 처리

Coroutine : 비동기 작업을 처리하기 위해 사용되는 경량 스레드

 

코루틴 빌더

runBlocking

현재 스레드를 차단하여 코루틴을 실행한다. 주로 main 함수나 test에서 사용

launch

새로운 Coroutine을 생성하고 결과를 반환하지 않는다. 주로 작업을 수행할 때 사용

async

새로운 Coroutine을 생성하고 결과를 Deferred 객체로 반환한다. 결과를 기다려야 할 때 사용

 

Dispatcher : Coroutine이 실행되는 스레드, 다양한 디스패처를 사용하여 코루틴의 실행 위치를 제어할 수 있다.

• Dispatchers.Main : UI와 상호작용하고 빠른 작업을 실행하기 위해서만 사용해야 한다.
• Dispatchers.IO : 메인 스레드 외부에서 디스크 또는 네트워크 I/O를 실행하도록 최적화되어 있다.
• Dispatchers.Default : CPU를 많이 사용하는 작업을 실행하는데 최적화되어 있다. (list sort, Json parsing..)

CoroutineScope

Coroutine이 실행되는 범위를 지정한다. Scope가 취소되면 그 안에서 실행되는 모든 Coroutine도 취소된다.

cancel( ) : Coroutine을 수동으로 취소 

withContext : 다른 디스패처에서 코루틴을 실행할 때 사용

fun main() = runBlocking {
    launch {
        withContext(Dispatchers.IO) {
            println("Dispather.IO에서 작업 중")
        }
    }
}

suspend 함수 : 코루틴 내에서 비동기 작업을 수행할 때 사용

---

병렬처리와 스레드 (Parallel Processing and Threads)

병렬처리

동시에 여러 작업을 수행하여 작업의 실행 시간을 단축하는 기법(멀티스레딩, 멀티 코어 CPU, 분산 시스템 등을 이용)

Kotlin에서는 스레드나 코루틴을 사용하여 병렬 처리를 구현할 수 있다.

 

suspend fun parallelLaunchDelay() = coroutineScope {
    launch {
        delay(1000) // 1. 실행흐름 밖으로 넘기고, 자기 할일
        println("print after 1 second same time. job1")
    }
    launch {
        delay(1000) // 2. 실행흐름 밖으로 넘기고, 자기 할일
        println("print after 1 second same time. job2")
    }
    println("print first") // 3. 위에 두개가 다 넘겨서 이쪽을 실행
    delay(2000) // 4. 두개 launch 밖의 코루틴 범위에서 기다림
    
}

fun main() = runBlocking<Unit> {
    parallelLaunchDelay()
}

• suspend : 코루틴 안에서만 호출될 수 있는 키워드.  coroutineScope, delay 도 suspend 함수이다. 코드의 실행흐름이 suspend 함수를 만나면 실행 제어흐름이 호출자한테로(밖으로) 넘어간다. 
• coroutineScope : 코루틴의 범위(scope) 를 나타낸다. 코루틴은 스코프에 어떤지에 따라서 예외를 전파시킬 수도, 아닐 수도 있다. 올바른 에러 전파 범위를 지정하기 위해서는 필수이다. 메서드를 리팩토링 하듯이 사용해야 한다고 생각한다.
• launch : 코루틴 builder라고 한다. 코루틴을 만들어 실행시키는 함수이며 응답값으로 Job 객체를 받을 수 있다. suspend 함수가 아니다.
• delay : 코루틴을 일시적으로 멈춘다. suspend 함수이다.
• runBlocking : suspend 세계와 블록킹 세계와 연결해주는 함수. runBlocking 을 호출한 스레드는 runBlocking 이 다 끝날때 까지 기다리게 된다. 

---

스레드 풀 (Thread Pool)

일정 수의 스레드를 미리 생성하여 관리하며, 작업이 요청될 때 이 스레드들을 할당하여 작업을 처리하도록 하는 구조이며작업이 완료되면 스레드는 종료되지 않고 풀에 다시 반환되어 다른 작업에 재사용된다.

특징

스레드 재사용

스레드 생성 및 종료에 따른 오버헤드를 줄일 수 있다.

 

작업 큐

스레드가 처리할 작업을 저장하고 스레드는 큐에서 작업을 꺼내어 실행한다.

 

스레드 수 제한

스레드 풀은 생성할 스레드의 수를 제한하여 시스템 자원의 낭비를 방지한다.

일반적으로 최소 스레드 수와 최대 스레드 수를 설정할 수 있다.

 

비동기 작업 처리

비동기 작업을 스레드 풀에 제출하여 멀티스레드 환경에서 동시에 작업을 처리할 수 있다.

fun main() = runBlocking {
    // 디스패처를 이용해 작업을 스레드 풀에서 실행
    val jobs = List(10) { i ->
        launch(Dispatchers.Default) {
            println("작업 $i이 ${Thread.currentThread().name}에서 실행")
            delay(1000) // 비동기 작업 시뮬레이션
        }
    }
    // 모든 작업이 완료될 때까지 대기
    jobs.forEach { it.join() }
    println("모든 작업 완료")
}

---

이벤트 큐 (Event Queue)

발생한 이벤트들을 순서대로 저장하고, 이벤트 루프에 의해 하나씩 꺼내어 처리하는 데이터 구조

---

이벤트 루프

이벤트가 발생시 호출되는 콜백 함수들을 콜백 큐에 전달하고 콜백 큐에 담겨있는 콜백 함수들을 콜스택으로 넘겨준다.

 

1. 이벤트 발생

2. 콜백 큐에 추가

• 비동기 작업이 완료되면 해당 작업의 콜백 함수나 결과가 코루틴의 suspend 함수로 처리되며, 큐에 추가 할 수 있다.
• 코루틴은 Deferred와 같은 비동기 결과 객체를 통해 작업 완료 후의 처리를 관리한다.

3. 이벤트 루프의 역할

• Kotlin에서는 명시적인 이벤트 루프가 존재하지 않지만, 코루틴의 동작이 이벤트 루프와 유사한 역할을 한다.
• CoroutineDispatcher와 Executor를 통해 비동기 작업을 스케줄링하고 실행한다..
• 코루틴은 launch, async, withContext 등의 빌더를 통해 비동기 작업을 정의하고 실행한다.

4. 콜백 큐에서 처리

• Kotlin의 코루틴은 기본적으로 스레드와 큐를 관리하여 비동기 작업을 처리합니다. CoroutineScope와 Dispatcher는 이러한 비동기 작업을 효율적으로 관리한다.
• runBlocking이나 GlobalScope.launch를 사용하여 코루틴을 실행하고, 작업 완료 후 결과를 처리한다.

fun main() {
    // 이벤트 큐 생성
    val callbackQueue = ConcurrentLinkedQueue<suspend () -> Unit>()

    // 이벤트 루프 실행
    val eventLoop = GlobalScope.launch {
        while (true) {
            val callback = callbackQueue.poll()
            if (callback != null) {
                callback()
            } else {
                delay(100) // 큐가 비어있으면 잠시 대기
            }
        }
    }

    // 이벤트 발생 시 콜백 큐에 추가
    GlobalScope.launch {
        delay(500)
        callbackQueue.offer {
            println("event 1 처리")
            delay(100)
        }
        
        delay(1000)
        callbackQueue.offer {
            println("event 2 처리")
            delay(100)
        }
    }

    // 메인 스레드가 종료되지 않도록 잠시 대기
    runBlocking {
        delay(3000)
    }
}

---

이벤트 핸들러 (Event Handler)

특정 이벤트가 발생했을 때 실행되는 코드를 정의한 함수 또는 메서드

특징 및 기능

• 특정 이벤트 처리: 이벤트 핸들러는 특정 이벤트가 발생했을 때 호출되어 해당 이벤트에 대한 적절한 동작 수행
• 콜백 함수: 이벤트 핸들러는 일반적으로 콜백 함수로 구현되며, 이벤트가 발생하면 자동으로 호출
• 비동기 처리: 이벤트 핸들러는 주로 비동기적으로 작동하여 이벤트 발생 시 비동기 작업 수행
• 유연성: 다양한 이벤트에 대해 각각의 핸들러를 정의할 수 있어 유연한 이벤트 처리 구조를 가질 수 있음

장점

• 모듈화: 이벤트 처리 코드를 별도의 핸들러로 분리함으로써 코드의 모듈화와 재사용성 향상
• 비동기 작업 처리: UI 작업이나 네트워크 요청 등의 비동기 작업을 효율적으로 처리
• 유지보수성: 이벤트와 관련된 코드를 한 곳에 모아 관리함으로써 코드의 유지보수성 좋아짐

단점

• 복잡성 증가: 이벤트 핸들러가 많아지면 코드의 복잡도가 증가
• 디버깅 어려움: 비동기적으로 실행되는 코드로 인해 디버깅이 어려울 수 있다.
• 메모리 누수: 잘못된 이벤트 핸들러 관리로 인해 메모리 누수가 발생할 수 있다.

---

ConcurrentLinkedQueue

멀티스레드 환경에서 안전하게 사용하기 위해 설계된 비차단 큐로, 연결 리스트를 기반으로 하여 요소를 관리하며, 여러 스레드가 동시에 큐에 접근해도 동시성 문제 없이 안전하게 작업을 수행할 수 있도록 해주는 자료 구조

 

fun main() {
    // ConcurrentLinkedQueue 생성
    val queue = ConcurrentLinkedQueue<Int>()

    // 큐에 요소 추가
    queue.offer(1)
    queue.offer(2)
    queue.offer(3)

    // 여러 스레드가 큐에 접근하여 요소를 추가 및 제거
    val threads = List(10) { i ->
        thread {
            // 요소 추가
            queue.offer(i)
            // 요소 제거
            val removedElement = queue.poll()
            println("Thread $i removed: $removedElement")
        }
    }

    // 모든 스레드가 종료될 때까지 기다림
    threads.forEach { it.join() }
}

---

채널 (Channel)

채널은 2개의 코루틴 사이를 연결할 수 있는 파이프와 같은 것이며, 임의의 데이터 스트림을 코루틴 사이에 공유할 수 있다. 만약 채널 내부의 용량이 다 찼는데 데이터를 채널에 보내려고 하면, 채널은 현재 코루틴을 일시 중단시키고 나중에 재게하게 된다.

장점

• 안전한 통신: 고루틴 간 안전한 데이터 교환 가능.
• 단순화된 동기화: 동기화 문제를 단순화.

단점

• 성능 오버헤드: 채널 사용으로 인한 성능 오버헤드 발생 가능.
• 복잡성: 복잡한 동시성 제어 시 채널 관리가 어려울 수 있음.

채널의 타입

1. Buffered

• 고정된 크기의 버퍼를 생성한다.
• 버퍼가 가득 차면 소비되기 전까지 새로운 데이터를 생산할 수 없다.

 

2. Rendezvous(Unbuffered)

• 아무 버퍼가 없는 랑데부 채녈 형태
• send() 호출은 receive()가 호출되기 전까지 항상 일시 중단되며 receive() 호출은 send()를 호출할 때까지 일시중단되게 된다.
• 랑데부 채널은 딜레이 시간과 관계없이 안정적인 동작 순서를 볼 수 있다.

3. Unlimited

• 버퍼의 용량 제한이 없고 버퍼의 용량은 필요에 따라 증가한다.
• send() 시에 일시 중단되는 일이 없지만 receive()시 버퍼가 비어있다면 일시 중단된다.

4. Conflated

• 송신된 값이 합쳐지는 채널
• 크기가 1인 고정 버퍼가 있는 채널이 생성되고 send()로 보낸 원소가 수신되기 전에 다른 값이 send() 되면 기존의 값을 덮어버린다. 이렇데 되면 기존의 원소값은 사라지게 되고 send()는 일시 중단되지 않는다.