코루틴 공부

 

Lazy = 큐에 올려놓고 start() 해야 동작할수있게

 

 

buffer() : collect 할때까지 기다리지 않고 다음꺼 미리 내놔 

-> 일감 있으면 일단 먼저 줘봐 !

 

 

conflate() : collect 할때 처리하지 못했던 값은 누락시킴 

 

collectLatest() : switchMap 같이 처리하다가 새로운값 들어오면 다시 시작 

-> 옆치기 들어오면 그것만 하고 하고 있는건 안해!

 

 

 

 

• 채널(channels): 코루틴 간에 데이터를 안전하게 보내고 받는 데 사용할 수 있는 파이프다.
• 작업자풀(worker pools): 많은 스레드에서 연산 집합의 처리를 나눌 수 있는 코루틴의 풀이다.
• 액터(actors): 채널과 코루틴을 사용하는 상태를 감싼 래퍼로 여러 쓰레드에서 상태를 안전하게 수정하는 메커니즘을 제공한다.
• 뮤텍스(mutexes): 크리티컬 존 영역을 정의해 한 번에 하나의 스레드만 실행할 수 있도록 하는 동기화 메커니즘. 크리티컬 존에 액세스하려는 코루틴은 이전 코루틴이 크리티컬 존을 빠져나올 때까지 일시 정지된다.
• 스레드 한정(Thread confinement): 코루틴의 실행을 제한해서 지정된 스레드에서만 실행하도록 하는 기능이다.
• 생성자(반복자 및 시퀀스: 필요에 따라 정보를 생성할 수 있고 새로운 정보가 필요하지 않을 때 일시 중단될 수 있는 데이터 소스다.)

 

 

Dispatcher 

1. Default는 코어 수에 비례하는 스레드 풀에서 수행합니다.
2. IO는 코어 수 보다 훨씬 많은 스레드를 가지는 스레드 풀입니다. IO 작업은 CPU를 덜 소모하기 때문입니다.
3. Unconfined는 어디에도 속하지 않습니다. 서스펜션 포인트를 지나면 바뀜
4. newSingleThreadContext는 항상 새로운 스레드를 만듭니다.

 

슈퍼바이저 스코프를 사용할 때 주의점은 무조건 자식 수준에서 예외를 핸들링 해야한다는 것입니다. 자식의 실패가 부모에게 전달되지 않기 때문에 자식 수준에서 예외를 처리해야합니다.

 

 

공유객체 문제

1. @Volatile 공유 변수에 사용

2. AtomicInteger 같은 스레드 안전한 자료 구조 이용

3. newSingleThreadContext 스레드 관리 

 

4. Mutex 사용

val mutex = Mutex()
var counter = 0

fun main() = runBlocking {
    withContext(Dispatchers.Default) {
        massiveRun {
            mutex.withLock {
                counter++
            }
        }
    }
    println("Counter = $counter")
}

 

5. 액터 : 액터가 독점적으로 자료를 가지며 그 자료를 다른 코루틴과 공유하지 않고 액터를 통해서만 접근하게 만듭니다.

sealed class CounterMsg
object IncCounter : CounterMsg()
class GetCounter(val response: CompletableDeferred<Int>) : CounterMsg()

fun CoroutineScope.counterActor() = actor<CounterMsg> {
    var counter = 0
    for (msg in channel) {
        when (msg) {
            is IncCounter -> counter++
            is GetCounter -> msg.response.complete(counter)
        }
    }
}

fun main() = runBlocking<Unit> {
    val counter = counterActor()
    withContext(Dispatchers.Default) {
        massiveRun {
            counter.send(IncCounter)
        }
    }

    val response = CompletableDeferred<Int>()
    counter.send(GetCounter(response))
    println("Counter = ${response.await()}")
    counter.close()
}

 

Flow : 비동기 스트림 ( 기본 콜드 스트림 ) 

 

flatMapConcat은 첫번째 요소에 대해서 플레트닝을 하고 나서 두번째 요소를 합니다.

flatMapMerge는 첫 요소의 프레트닝을 시작하며 이어 다음 요소의 플레트닝을 시작합니다.

 

 

flatMapLatest는 다음 요소의 플레트닝을 시작하며 이전에 진행 중이던 플레트닝을 취소합니다. (switchMap)

 

 

catch 연산자는 업스트림(catch 연산자를 쓰기 전의 코드)에만 영향을 미치고 다운스트림에는 영향을 미치지 않습니다. 이를 catch 투명성이라 합니다. 

flow collect 할때 예외처리는  .catch 로 하는걸 추천!!

-> 아니 .onCompletion 으로 하면 finally 처리와 예외처리 까지 한꺼번에 할수있어 = 이게 더 좋아

 

 

 

launchIn 을 사용하여 새로운 코루틴으로 돌아갈수있게 (collect 를 하면 그 플로우가 끝날때까지 잡고 있음 ) 

 

 

 

Channel :  채널은 일종의 파이프입니다. 송신측에서 채널에 send로 데이터를 전달하고 수신 측에서 채널을 통해 receive 받습니다. 

 

Q. 언제 채널플로우를 사용하는지?- 플로우는 채널플로우의 일부분이다.  (채널플로우가 더 무겁다) - 플로우 내부에서 비동기작업을 할때 채널플로우 사용하고, - 만약 외부로 데이터를 전달하는 경우엔 callbackFlow + awaitClose{ } 를 사용한다 

 

팬 아웃

- 여러 코루틴이 동시에 채널을 구독할 수 있습니다.

팬 인

 - 팬 인은 반대로 생산자가 많은 것입니다.