코루틴과 채널
코루틴과 채널
네트워크 요청을 할 때 스레드를 막지도 않고 콜백 지옥에 빠지지도 않는 방법을 찾고 있나요? 그 해답이 바로 코루틴(coroutines)과 채널(channels)이에요. 이 튜토리얼에서는 IntelliJ IDEA에서 코루틴을 활용해 GitHub 기여자 목록을 비동기로 불러오는 예제를 따라가며, 왜 이렇게 쓰는지부터 채널로 코루틴 사이에 데이터를 주고받는 방법까지 직접 구현해 볼게요.
출처: Kotlin 공식 문서
본문
곧 예정된 업데이트가 이 튜토리얼을 개정할 예정이에요. 그동안 최신 코루틴 시작 가이드는 코루틴 기초(Coroutines basics)를 참고하세요.
이 튜토리얼에서는 IntelliJ IDEA에서 코루틴을 사용해, 기반 스레드를 막거나 콜백을 사용하지 않고 네트워크 요청을 수행하는 방법을 배워요.
코루틴에 대한 사전 지식은 필요 없지만, 기본적인 Kotlin 문법에는 익숙해야 해요.
배우게 될 내용:
- 중단 함수(suspending functions)를 사용해 네트워크 요청을 수행하는 이유와 방법
- 코루틴을 사용해 요청을 동시에(concurrently) 보내는 방법
- 채널을 사용해 서로 다른 코루틴 간에 정보를 공유하는 방법
네트워크 요청에는 Retrofit 라이브러리가 필요하지만, 이 튜토리얼에서 보여주는 방식은 코루틴을 지원하는 다른 라이브러리에서도 비슷하게 동작해요.
프로젝트 저장소의
solutions브랜치에서 모든 태스크의 해결책을 찾을 수 있어요.
시작하기 전에
- 최신 버전의 IntelliJ IDEA를 다운로드해서 설치해요.
- Welcome 화면에서 Get from VCS를 선택하거나 File | New | Project from Version Control을 선택해 프로젝트 템플릿을 클론해요.
명령줄에서도 클론할 수 있어요.
git clone https://github.com/kotlin-hands-on/intro-coroutines
GitHub 개발자 토큰 생성
프로젝트에서 GitHub API를 사용할 거예요. 접근 권한을 얻으려면 GitHub 계정 이름과 비밀번호 또는 토큰을 제공해야 해요. 2단계 인증(two-factor authentication)을 활성화했다면 토큰으로 충분해요.
계정으로 GitHub API를 사용할 새 GitHub 토큰을 생성해요.
- 토큰 이름을 지정해요. 예:
coroutines-tutorial - 어떤 scope도 선택하지 마세요. 페이지 하단의 Generate token을 클릭해요.
- 생성된 토큰을 복사해요.
코드 실행
프로그램은 주어진 조직(기본값 "kotlin") 아래 모든 저장소의 기여자(contributors)를 로드해요. 나중에 사용자들을 기여 횟수로 정렬하는 로직을 추가할 거예요.
src/contributors/main.kt파일을 열고main()함수를 실행해요. 다음 창이 보일 거예요.
글꼴이 너무 작다면 main() 함수에서 setDefaultFontSize(18f) 값을 바꿔 조정해요.
- 해당 필드에 GitHub 사용자 이름과 토큰(또는 비밀번호)을 입력해요.
- Variant 드롭다운 메뉴에서 BLOCKING 옵션이 선택돼 있는지 확인해요.
- Load contributors를 클릭해요. 잠시 UI가 멈춘(freeze) 후 기여자 목록이 표시될 거예요.
- 프로그램 출력을 열어 데이터가 로드됐는지 확인해요. 기여자 목록은 각 요청이 성공한 후 기록돼요.
이 로직은 blocking 요청이나 콜백을 사용해 구현할 수 있어요. 이 해결책들을 코루틴을 사용하는 해결책과 비교하고, 채널로 서로 다른 코루틴 사이에 정보를 공유하는 방법도 살펴볼 거예요.
Blocking 요청
GitHub에 HTTP 요청을 수행하려면 Retrofit 라이브러리를 사용할 거예요. Retrofit을 사용하면 주어진 조직 아래의 저장소 목록과 각 저장소의 기여자 목록을 요청할 수 있어요.
interface GitHubService {
@GET("orgs/{org}/repos?per_page=100")
fun getOrgReposCall(
@Path("org") org: String
): Call<List<Repo>>
@GET("repos/{owner}/{repo}/contributors?per_page=100")
fun getRepoContributorsCall(
@Path("owner") owner: String,
@Path("repo") repo: String
): Call<List<User>>
}
이 API는 loadContributorsBlocking() 함수가 주어진 조직의 기여자 목록을 가져오는 데 사용해요.
src/tasks/Request1Blocking.kt를 열어 구현을 확인해요.
fun loadContributorsBlocking(
service: GitHubService,
req: RequestData
): List<User> {
val repos = service
.getOrgReposCall(req.org) // #1
.execute() // #2
.also { logRepos(req, it) } // #3
.body() ?: emptyList() // #4
return repos.flatMap { repo ->
service
.getRepoContributorsCall(req.org, repo.name) // #1
.execute() // #2
.also { logUsers(repo, it) } // #3
.bodyList() // #4
}.aggregate()
}
- 먼저 주어진 조직 아래의 저장소 목록을 얻어
repos리스트에 저장해요. 그런 다음 각 저장소에 대해 기여자 목록을 요청하고, 모든 목록을 하나의 최종 기여자 목록으로 병합해요. getOrgReposCall()과getRepoContributorsCall()은 둘 다*Call클래스의 인스턴스를 반환해요(#1). 이 시점에는 요청이 전송되지 않아요.- 그다음
*Call.execute()를 호출해 요청을 수행해요(#2).execute()는 기반 스레드를 막는 동기 호출이에요. - 응답을 받으면 특정
logRepos()와logUsers()함수를 호출해 결과를 기록해요(#3). HTTP 응답에 오류가 있으면 여기서 그 오류가 기록돼요. - 마지막으로 필요한 데이터가 담긴 응답의 본문(body)을 가져와요. 이 튜토리얼에서는 오류가 있을 때 결과로 빈 리스트를 사용하고, 해당 오류를 기록해요(
#4). .body() ?: emptyList()를 반복하지 않기 위해 확장 함수bodyList()를 선언해요.
fun <T> Response<List<T>>.bodyList(): List<T> {
return body() ?: emptyList()
}
- 프로그램을 다시 실행하고 IntelliJ IDEA의 시스템 출력을 살펴봐요. 대략 다음과 같아야 해요.
1770 [AWT-EventQueue-0] INFO Contributors - kotlin: loaded 40 repos
2025 [AWT-EventQueue-0] INFO Contributors - kotlin-examples: loaded 23 contributors
2229 [AWT-EventQueue-0] INFO Contributors - kotlin-koans: loaded 45 contributors
...
- 각 줄의 첫 번째 항목은 프로그램 시작 후 지난 밀리초 수이고, 그다음 대괄호 안이 스레드 이름이에요. 로딩 요청이 어느 스레드에서 호출되는지 볼 수 있죠.
- 각 줄의 마지막 항목은 실제 메시지예요. 로드된 저장소나 기여자가 몇 개인지에요.
이 로그 출력은 모든 결과가 main 스레드에서 기록됐음을 보여줘요. BLOCKING 옵션으로 코드를 실행하면 창이 멈추고 로딩이 끝날 때까지 입력에 반응하지 않아요. 모든 요청은 loadContributorsBlocking()이 호출되는 것과 같은 스레드에서 실행되는데, 이는 main UI 스레드예요 (Swing에서는 AWT 이벤트 디스패칭 스레드). 이 main 스레드가 막히고, 그래서 UI가 얼어붙는 거예요.
기여자 목록이 로드된 후에 결과가 갱신돼요.
src/contributors/Contributors.kt에서loadContributors()함수를 찾아 기여자가 어떻게 로드될지 선택하고,loadContributorsBlocking()이 어떻게 호출되는지 살펴봐요.
when (getSelectedVariant()) {
BLOCKING -> { // Blocking UI thread
val users = loadContributorsBlocking(service, req)
updateResults(users, startTime)
}
}
updateResults()호출은loadContributorsBlocking()호출 바로 다음에 와요.updateResults()는 UI를 갱신하므로 항상 UI 스레드에서 호출돼야 해요.loadContributorsBlocking()도 UI 스레드에서 호출되므로 UI 스레드가 막히고 UI가 얼어붙어요.
태스크 1
첫 번째 태스크는 작업 영역에 익숙해지도록 도와줘요. 현재 각 기여자의 이름은 참여한 프로젝트마다 여러 번 반복돼요. aggregate() 함수를 구현해서 각 기여자가 한 번만 추가되도록 사용자들을 결합해요. User.contributions 속성은 주어진 사용자가 모든 프로젝트에 기여한 총 횟수를 담아야 해요. 결과 리스트는 기여 횟수에 따라 내림차순으로 정렬돼야 해요.
src/tasks/Aggregation.kt를 열고 List<User>.aggregate() 함수를 구현해요. 사용자는 총 기여 횟수로 정렬돼야 해요.
해당 테스트 파일 test/tasks/AggregationKtTest.kt가 기대 결과의 예를 보여줘요.
IntelliJ IDEA 단축키
Ctrl+Shift+T/⇧ ⌘ T로 소스 코드와 테스트 클래스 사이를 자동으로 오갈 수 있어요.
이 태스크를 구현한 후 "kotlin" 조직의 결과 리스트는 대략 다음과 같아야 해요.
태스크 1의 해결 방법
- 로그인으로 사용자를 그룹화하려면
groupBy()를 사용해요. 이 함수는 로그인에서 서로 다른 저장소에서 이 로그인을 가진 사용자의 모든 등장으로 이어지는 맵을 반환해요. - 각 맵 항목에 대해 각 사용자의 총 기여 횟수를 세고, 주어진 이름과 총 기여 횟수로
User클래스의 새 인스턴스를 만들어요. - 결과 리스트를 내림차순으로 정렬해요.
fun List<User>.aggregate(): List<User> =
groupBy { it.login }
.map { (login, group) -> User(login, group.sumOf { it.contributions }) }
.sortedByDescending { it.contributions }
대안으로 groupBy() 대신 groupingBy() 함수를 사용할 수도 있어요.
콜백
이전 해결책은 동작하지만 스레드를 막아서 UI를 얼리죠. 이를 피하는 전통적인 접근 방식은 콜백(callbacks)을 사용하는 거예요.
연산 완료 직후 호출해야 하는 코드를 호출하는 대신, 이를 별도의 콜백(흔히 람다)으로 추출하고 그 람다를 호출자에게 전달해서 나중에 호출되게 할 수 있어요.
UI를 반응적으로 만들려면 전체 계산을 별도의 스레드로 옮기거나, blocking 호출 대신 콜백을 사용하는 Retrofit API로 전환할 수 있어요.
백그라운드 스레드 사용
src/tasks/Request2Background.kt를 열고 구현을 살펴봐요. 먼저 전체 계산을 다른 스레드로 옮겨요.thread()함수가 새 스레드를 시작하죠.
thread {
loadContributorsBlocking(service, req)
}
이제 모든 로딩이 별도의 스레드로 옮겨졌으므로 main 스레드는 자유롭고 다른 작업이 차지할 수 있어요.
loadContributorsBackground()함수의 시그니처가 바뀌어요. 로딩이 모두 끝난 뒤 호출할updateResults()콜백을 마지막 인자로 받죠.
fun loadContributorsBackground(
service: GitHubService, req: RequestData,
updateResults: (List<User>) -> Unit
)
- 이제
loadContributorsBackground()가 호출되면updateResults()호출은 이전처럼 바로 다음이 아니라 콜백 안에서 이뤄져요.
loadContributorsBackground(service, req) { users ->
SwingUtilities.invokeLater {
updateResults(users, startTime)
}
}
SwingUtilities.invokeLater를 호출해서 결과를 갱신하는 updateResults() 호출이 main UI 스레드(AWT 이벤트 디스패칭 스레드)에서 일어나도록 보장해요.
하지만 BACKGROUND 옵션으로 기여자를 로드해 보면, 목록이 갱신돼도 UI에는 아무 변화가 없는 걸 볼 수 있어요.
태스크 2
src/tasks/Request2Background.kt의 loadContributorsBackground() 함수를 고쳐서 결과 리스트가 UI에 표시되게 해요.
태스크 2의 해결 방법
기여자를 로드해 보면 로그에는 기여자가 로드됐지만 결과가 표시되지 않는 걸 볼 수 있어요. 이를 고치려면 결과 사용자 리스트에 대해 updateResults()를 호출해요.
thread {
updateResults(loadContributorsBlocking(service, req))
}
콜백으로 전달된 로직을 명시적으로 호출해야 해요. 그렇게 하지 않으면 아무 일도 일어나지 않아요.
Retrofit 콜백 API 사용
이전 해결책에서는 전체 로딩 로직이 백그라운드 스레드로 옮겨지지만, 여전히 자원을 최적으로 쓰는 건 아니에요. 모든 로딩 요청은 순차적으로 진행되고 스레드는 로딩 결과를 기다리는 동안 막혀 있는데, 그 동안 다른 작업이 스레드를 차지할 수 있었죠. 구체적으로 스레드가 다른 요청 로딩을 시작해서 전체 결과를 더 일찍 받을 수 있었어요.
각 저장소의 데이터 처리는 로딩과 응답 처리 두 부분으로 나눠야 해요. 두 번째 처리 부분은 콜백으로 추출돼야 하죠.
그러면 각 저장소의 로딩을 이전 저장소의 결과를 받기 전에(그리고 해당 콜백이 호출되기 전에) 시작할 수 있어요.
Retrofit 콜백 API가 이를 도와줘요. Call.enqueue() 함수가 HTTP 요청을 시작하고 콜백을 인자로 받아요. 이 콜백에서 각 요청 후 무엇을 해야 할지 지정하면 돼요.
src/tasks/Request3Callbacks.kt를 열고 이 API를 사용하는 loadContributorsCallbacks() 구현을 살펴봐요.
fun loadContributorsCallbacks(
service: GitHubService, req: RequestData,
updateResults: (List<User>) -> Unit
) {
service.getOrgReposCall(req.org).onResponse { responseRepos -> // #1
logRepos(req, responseRepos)
val repos = responseRepos.bodyList()
val allUsers = mutableListOf<User>()
for (repo in repos) {
service.getRepoContributorsCall(req.org, repo.name)
.onResponse { responseUsers -> // #2
logUsers(repo, responseUsers)
val users = responseUsers.bodyList()
allUsers += users
}
}
}
// TODO: 이 코드는 왜 동작하지 않을까? 어떻게 고칠까?
updateResults(allUsers.aggregate())
}
- 편의상 이 코드 조각은 같은 파일에 선언된
onResponse()확장 함수를 사용해요. 객체 표현식 대신 람다를 인자로 받죠. - 응답 처리 로직은 콜백으로 추출돼요. 해당 람다는
#1과#2줄에서 시작해요.
하지만 제공된 해결책은 동작하지 않아요. 프로그램을 실행하고 CALLBACKS 옵션으로 기여자를 로드하면 아무것도 표시되지 않아요. 그런데 Request3CallbacksKtTest의 테스트는 성공적으로 통과했다는 결과를 즉시 반환해요.
주어진 코드가 왜 기대대로 동작하지 않는지 생각해 보고 고쳐 보거나, 아래 해결책을 확인해 보세요.
태스크 3 (선택)
src/tasks/Request3Callbacks.kt 파일의 코드를 다시 작성해서 로드된 기여자 목록이 표시되게 해요.
태스크 3의 첫 번째 시도 해결책
현재 해결책에서는 많은 요청이 동시에 시작돼서 전체 로딩 시간이 줄어들어요. 하지만 결과가 로드되지 않아요. updateResults() 콜백이 모든 로딩 요청이 시작된 직후, allUsers 리스트가 데이터로 채워지기 전에 호출되기 때문이에요.
다음과 같은 변경으로 고쳐 볼 수 있어요.
val allUsers = mutableListOf<User>()
for ((index, repo) in repos.withIndex()) { // #1
service.getRepoContributorsCall(req.org, repo.name)
.onResponse { responseUsers ->
logUsers(repo, responseUsers)
val users = responseUsers.bodyList()
allUsers += users
if (index == repos.lastIndex) { // #2
updateResults(allUsers.aggregate())
}
}
}
- 먼저 인덱스와 함께 저장소 리스트를 순회해요(
#1). - 그다음 각 콜백에서 마지막 반복인지 확인해요(
#2). - 마지막이면 결과를 갱신해요.
하지만 이 코드도 목표를 달성하지 못해요. 직접 답을 찾아 보거나 아래 해결책을 확인해요.
태스크 3의 두 번째 시도 해결책
로딩 요청이 동시에 시작되므로 마지막 요청의 결과가 마지막에 온다는 보장이 없어요. 결과는 어떤 순서로든 올 수 있죠.
따라서 완료 조건으로 현재 인덱스를 lastIndex와 비교하면 일부 저장소의 결과를 잃을 위험이 있어요.
마지막 저장소를 처리하는 요청이 일부 이전 요청보다 빨리 반환되면(그럴 가능성이 높죠), 시간이 더 걸리는 요청들의 결과가 모두 사라져요.
이를 고치는 한 가지 방법은 인덱스를 도입해 모든 저장소가 이미 처리됐는지 확인하는 거예요.
val allUsers = Collections.synchronizedList(mutableListOf<User>())
val numberOfProcessed = AtomicInteger()
for (repo in repos) {
service.getRepoContributorsCall(req.org, repo.name)
.onResponse { responseUsers ->
logUsers(repo, responseUsers)
val users = responseUsers.bodyList()
allUsers += users
if (numberOfProcessed.incrementAndGet() == repos.size) {
updateResults(allUsers.aggregate())
}
}
}
이 코드는 동기화된 리스트와 AtomicInteger()를 사용해요. 일반적으로 getRepoContributors() 요청을 처리하는 서로 다른 콜백이 항상 같은 스레드에서 호출된다는 보장이 없기 때문이에요.
태스크 3의 세 번째 시도 해결책
더 나은 해결책은 CountDownLatch 클래스를 사용하는 거예요. 저장소 수로 초기화된 카운터를 저장하고, 각 저장소를 처리한 후 이 카운터를 감소시켜요. 그런 다음 래치가 0까지 카운트다운될 때까지 기다렸다가 결과를 갱신해요.
val countDownLatch = CountDownLatch(repos.size)
for (repo in repos) {
service.getRepoContributorsCall(req.org, repo.name)
.onResponse { responseUsers ->
// 저장소 처리
countDownLatch.countDown()
}
}
countDownLatch.await()
updateResults(allUsers.aggregate())
그러면 결과가 main 스레드에서 갱신돼요. 이는 로직을 자식 스레드에 위임하는 것보다 더 직접적이에요.
이 세 가지 해결책 시도를 살펴본 뒤, 콜백으로 올바른 코드를 작성하는 것은 사소하지 않고 오류가 발생하기 쉬우며, 특히 여러 기반 스레드와 동기화가 관련될 때 그렇다는 걸 알 수 있어요.
추가 연습으로 RxJava 라이브러리로 반응형 접근 방식을 사용해 같은 로직을 구현할 수 있어요. RxJava 사용에 필요한 모든 의존성과 해결책은 별도의
rx브랜치에서 찾을 수 있어요. 정확한 비교를 위해 이 튜토리얼의 Rx 버전을 완료하고 구현하거나 확인하는 것도 가능해요.
중단 함수
중단 함수(suspending functions)를 사용해 같은 로직을 구현할 수 있어요. Call<List<Repo>>를 반환하는 대신 API 호출을 다음과 같이 중단 함수로 정의해요.
interface GitHubService {
@GET("orgs/{org}/repos?per_page=100")
suspend fun getOrgRepos(
@Path("org") org: String
): List<Repo>
}
getOrgRepos()는suspend함수로 정의돼요. 중단 함수로 요청을 수행하면 기반 스레드가 막히지 않아요. 이것이 어떻게 동작하는지에 대한 자세한 내용은 나중 섹션에서 나와요.getOrgRepos()는Call을 반환하는 대신 결과를 직접 반환해요. 결과가 실패하면 예외가 던져져요.
또는 Retrofit은 결과를 Response로 감싼 형태로 반환하는 것도 허용해요. 이 경우 결과 본문이 제공되고 오류를 수동으로 확인할 수 있어요. 이 튜토리얼은 Response를 반환하는 버전을 사용해요.
src/contributors/GitHubService.kt에서 GitHubService 인터페이스에 다음 선언들을 추가해요.
interface GitHubService {
// getOrgReposCall & getRepoContributorsCall 선언
@GET("orgs/{org}/repos?per_page=100")
suspend fun getOrgRepos(
@Path("org") org: String
): Response<List<Repo>>
@GET("repos/{owner}/{repo}/contributors?per_page=100")
suspend fun getRepoContributors(
@Path("owner") owner: String,
@Path("repo") repo: String
): Response<List<User>>
}
태스크 4
기여자를 로드하는 함수의 코드를 바꿔서 두 개의 새 중단 함수 getOrgRepos()와 getRepoContributors()를 사용하게 만드는 것이 태스크예요. 새 loadContributorsSuspend() 함수는 새 API를 사용하기 위해 suspend로 표시돼요.
중단 함수는 어디에서나 호출할 수 없어요.
loadContributorsBlocking()에서 중단 함수를 호출하면 "Suspend function 'getOrgRepos' should be called only from a coroutine or another suspend function"이라는 메시지와 함께 오류가 발생해요.
src/tasks/Request1Blocking.kt에 정의된loadContributorsBlocking()의 구현을src/tasks/Request4Suspend.kt에 정의된loadContributorsSuspend()에 복사해요.Call을 반환하는 함수 대신 새 중단 함수가 사용되도록 코드를 수정해요.- SUSPEND 옵션을 선택해 프로그램을 실행하고, GitHub 요청이 수행되는 동안 UI가 여전히 반응적인지 확인해요.
태스크 4의 해결 방법
.getOrgReposCall(req.org).execute()를 .getOrgRepos(req.org)로 바꾸고, 두 번째 "contributors" 요청에도 같은 교체를 반복해요.
suspend fun loadContributorsSuspend(service: GitHubService, req: RequestData): List<User> {
val repos = service
.getOrgRepos(req.org)
.also { logRepos(req, it) }
.bodyList()
return repos.flatMap { repo ->
service.getRepoContributors(req.org, repo.name)
.also { logUsers(repo, it) }
.bodyList()
}.aggregate()
}
loadContributorsSuspend()는suspend함수로 정의돼야 해요.- 이전에
Response를 반환했던execute를 더는 호출할 필요가 없어요. 이제 API 함수가Response를 직접 반환하거든요. 이 세부 사항은 Retrofit 라이브러리에 특화된 것임을 기억하세요. 다른 라이브러리에서는 API가 다르지만 개념은 같아요.
코루틴
중단 함수가 있는 코드는 "blocking" 버전과 비슷해 보여요. blocking 버전과의 주요 차이는 스레드를 막는 대신 코루틴이 **중단(suspend)**된다는 점이에요.
block -> suspend
thread -> coroutine
코루틴은 종종 경량 스레드(lightweight threads)라고 불려요. 스레드에서 코드를 실행하는 것과 비슷하게 코루틴에서도 코드를 실행할 수 있기 때문이죠. 이전에 블로킹이었던(그래서 피해야 했던) 연산은 이제 코루틴을 중단하는 것으로 대체될 수 있어요.
새 코루틴 시작
src/contributors/Contributors.kt에서 loadContributorsSuspend()가 어떻게 사용되는지 보면, launch 안에서 호출되는 걸 알 수 있어요. launch는 람다를 인자로 받는 라이브러리 함수예요.
launch {
val users = loadContributorsSuspend(req)
updateResults(users, startTime)
}
여기서 launch는 데이터를 로드하고 결과를 보여주는 새 계산을 시작해요. 이 계산은 중단 가능한데, 네트워크 요청을 수행할 때 중단되고 기반 스레드를 해제해요. 네트워크 요청이 결과를 반환하면 계산이 재개돼요.
이런 중단 가능한 계산을 **코루틴(coroutine)**이라고 불러요. 그래서 이 경우 launch는 데이터 로드와 결과 표시를 담당하는 새 코루틴을 시작해요.
코루틴은 스레드 위에서 실행되고 중단될 수 있어요. 코루틴이 중단되면 해당 계산은 일시 정지되어 스레드에서 제거되고 메모리에 저장돼요. 그동안 스레드는 다른 작업이 차지할 수 있어요.
계산이 계속될 준비가 되면 스레드(반드시 같은 스레드일 필요는 없어요)로 복귀해요.
loadContributorsSuspend() 예시에서 각 "contributors" 요청은 이제 중단 메커니즘으로 결과를 기다려요. 먼저 새 요청이 전송되고, 그다음 응답을 기다리는 동안 launch 함수가 시작한 전체 "load contributors" 코루틴이 중단돼요.
코루틴은 해당 응답을 받은 후에만 재개돼요.
응답을 기다리는 동안 스레드는 다른 작업이 차지할 수 있어요. 모든 요청이 main UI 스레드에서 일어나는데도 UI는 반응적으로 유지돼요.
- SUSPEND 옵션으로 프로그램을 실행해요. 로그는 모든 요청이 main UI 스레드에서 전송됐음을 확인해 줘요.
2538 [AWT-EventQueue-0 @coroutine#1] INFO Contributors - kotlin: loaded 30 repos
2729 [AWT-EventQueue-0 @coroutine#1] INFO Contributors - ts2kt: loaded 11 contributors
3029 [AWT-EventQueue-0 @coroutine#1] INFO Contributors - kotlin-koans: loaded 45 contributors
...
11252 [AWT-EventQueue-0 @coroutine#1] INFO Contributors - kotlin-coroutines-workshop: loaded 1 contributors
- 로그는 해당 코드가 어느 코루틴에서 실행되는지 보여줄 수 있어요. 활성화하려면 Run | Edit configurations를 열고
-Dkotlinx.coroutines.debugVM 옵션을 추가해요.
이 옵션으로 main()을 실행하면 코루틴 이름이 스레드 이름에 붙어요. 모든 Kotlin 파일 실행용 템플릿을 수정해 이 옵션을 기본으로 활성화할 수도 있어요.
이제 모든 코드는 하나의 코루틴, 위에서 언급한 "load contributors" 코루틴(@coroutine#1)에서 실행돼요. 결과를 기다리는 동안 코드가 순차적으로 작성돼 있으므로 스레드를 다른 요청 전송에 재사용해서는 안 돼요. 이전 결과를 받아야만 새 요청이 전송돼요.
중단 함수는 스레드를 공정하게 다루고 "기다림"을 위해 스레드를 막지 않아요. 하지만 이것만으로는 아직 동시성(concurrency)이 생기지 않아요.
동시성
Kotlin 코루틴은 스레드보다 자원 소모가 훨씬 적어요. 비동기로 새 계산을 시작하고 싶을 때마다 새 코루틴을 만들면 돼요.
새 코루틴을 시작하려면 주요 코루틴 빌더 중 하나인 launch, async, runBlocking을 사용해요. 서로 다른 라이브러리가 추가 코루틴 빌더를 정의할 수 있어요.
async는 새 코루틴을 시작하고 Deferred 객체를 반환해요. Deferred는 다른 이름으로는 Future나 Promise라고도 알려진 개념을 나타내요. 계산을 저장하지만 최종 결과를 얻는 순간을 미루고, 미래 어느 시점에 결과를 약속해요.
async와 launch의 주요 차이는 launch가 특정 결과를 반환할 것으로 기대하지 않는 계산을 시작하는 데 사용된다는 점이에요. launch는 코루틴을 나타내는 Job을 반환해요. Job.join()을 호출해 완료될 때까지 기다릴 수 있어요.
Deferred는 Job을 확장하는 제네릭 타입이에요. async 호출은 람다가 반환하는 것에 따라 Deferred<Int>나 Deferred<CustomType>을 반환할 수 있어요 (람다 안의 마지막 표현식이 결과예요).
코루틴의 결과를 얻으려면 Deferred 인스턴스에서 await()를 호출할 수 있어요. 결과를 기다리는 동안 이 await()가 호출된 코루틴은 중단돼요.
import kotlinx.coroutines.*
fun main() = runBlocking {
val deferred: Deferred<Int> = async {
loadData()
}
println("waiting...")
println(deferred.await())
}
suspend fun loadData(): Int {
println("loading...")
delay(1000L)
println("loaded!")
return 42
}
runBlocking은 일반 함수와 중단 함수, 또는 블로킹 세계와 논블로킹 세계 사이의 다리로 사용돼요. 최상위 main 코루틴을 시작하는 어댑터처럼 동작하죠. 주로 main() 함수와 테스트에서 사용되도록 만들어졌어요.
더 나은 이해를 위해 이 비디오를 시청해 보세요.
deferred 객체 목록이 있으면 awaitAll()을 호출해 모두의 결과를 기다릴 수 있어요.
import kotlinx.coroutines.*
fun main() = runBlocking {
val deferreds: List<Deferred<Int>> = (1..3).map {
async {
delay(1000L * it)
println("Loading $it")
it
}
}
val sum = deferreds.awaitAll().sum()
println("$sum")
}
각 "contributors" 요청이 새 코루틴에서 시작되면 모든 요청이 비동기로 시작돼요. 이전 요청의 결과를 받기 전에 새 요청을 보낼 수 있죠.
전체 로딩 시간은 CALLBACKS 버전과 대략 같지만 콜백이 필요 없어요. 게다가 async는 코드에서 어떤 부분이 동시에 실행되는지 명시적으로 강조해요.
태스크 5
Request5Concurrent.kt 파일에서 이전 loadContributorsSuspend() 함수를 사용해 loadContributorsConcurrent() 함수를 구현해요.
태스크 5의 힌트
코루틴 스코프 안에서만 새 코루틴을 시작할 수 있어요. loadContributorsSuspend()의 내용을 coroutineScope 호출에 복사해서 그 안에서 async 함수를 호출할 수 있게 해요.
suspend fun loadContributorsConcurrent(
service: GitHubService,
req: RequestData
): List<User> = coroutineScope {
// ...
}
해결책을 다음 구조에 기반해요.
val deferreds: List<Deferred<List<User>>> = repos.map { repo ->
async {
// 각 repo의 기여자 로드
}
}
deferreds.awaitAll() // List<List<User>>
태스크 5의 해결 방법
각 "contributors" 요청을 async로 감싸서 저장소 수만큼 코루틴을 만들어요. async는 Deferred<List<User>>를 반환해요. 새 코루틴을 만드는 것은 자원 소모가 크지 않으므로 필요한 만큼 많이 만들 수 있어서 문제가 되지 않아요.
map결과가 이제 리스트의 리스트가 아니라Deferred객체의 리스트이므로 더는flatMap을 사용할 수 없어요.awaitAll()이List<List<User>>를 반환하므로 결과를 얻으려면flatten().aggregate()을 호출해요.
suspend fun loadContributorsConcurrent(
service: GitHubService,
req: RequestData
): List<User> = coroutineScope {
val repos = service
.getOrgRepos(req.org)
.also { logRepos(req, it) }
.bodyList()
val deferreds: List<Deferred<List<User>>> = repos.map { repo ->
async {
service.getRepoContributors(req.org, repo.name)
.also { logUsers(repo, it) }
.bodyList()
}
}
deferreds.awaitAll().flatten().aggregate()
}
- 코드를 실행하고 로그를 확인해요. 아직 멀티스레딩을 사용하지 않았으므로 모든 코루틴이 여전히 main UI 스레드에서 실행되지만, 코루틴을 동시에 실행하는 이점은 이미 볼 수 있어요.
- "contributors" 코루틴이 공용 스레드 풀의 서로 다른 스레드에서 실행되도록 이 코드를 바꾸려면
async함수의 컨텍스트 인자로Dispatchers.Default를 지정해요.
async(Dispatchers.Default) { }
CoroutineDispatcher는 해당 코루틴이 어떤 스레드에서 실행돼야 하는지 결정해요. 인자로 지정하지 않으면async는 바깥 스코프의 디스패처를 사용해요.Dispatchers.Default는 JVM의 공유 스레드 풀을 나타내요. 이 풀은 병렬 실행 수단을 제공해요. 사용 가능한 CPU 코어 수만큼의 스레드로 구성되지만, 코어가 하나뿐이어도 스레드 두 개는 갖게 돼요.loadContributorsConcurrent()함수의 코드를 수정해서 공용 스레드 풀의 서로 다른 스레드에서 새 코루틴을 시작하게 해요. 요청을 보내기 전에 추가 로깅도 넣어요.
async(Dispatchers.Default) {
log("starting loading for ${repo.name}")
service.getRepoContributors(req.org, repo.name)
.also { logUsers(repo, it) }
.bodyList()
}
- 프로그램을 다시 실행해요. 로그에서 각 코루틴이 스레드 풀의 한 스레드에서 시작되고 다른 스레드에서 재개될 수 있음을 볼 수 있어요.
1946 [DefaultDispatcher-worker-2 @coroutine#4] INFO Contributors - starting loading for kotlin-koans
1946 [DefaultDispatcher-worker-3 @coroutine#5] INFO Contributors - starting loading for dokka
1946 [DefaultDispatcher-worker-1 @coroutine#3] INFO Contributors - starting loading for ts2kt
...
2178 [DefaultDispatcher-worker-1 @coroutine#4] INFO Contributors - kotlin-koans: loaded 45 contributors
2569 [DefaultDispatcher-worker-1 @coroutine#5] INFO Contributors - dokka: loaded 36 contributors
2821 [DefaultDispatcher-worker-2 @coroutine#3] INFO Contributors - ts2kt: loaded 11 contributors
예를 들어 이 로그 발췌에서 coroutine#4는 worker-2 스레드에서 시작되고 worker-1 스레드에서 계속됩니다.
src/contributors/Contributors.kt에서 CONCURRENT 옵션의 구현을 확인해요.
- 코루틴을 main UI 스레드에서만 실행하려면 인자로
Dispatchers.Main을 지정해요.
launch(Dispatchers.Main) {
updateResults()
}
- main 스레드가 바쁠 때 그 위에서 새 코루틴을 시작하면 코루틴은 중단되어 이 스레드에서 실행되도록 예약돼요. 코루틴은 스레드가 자유로워질 때만 재개돼요.
- 각 종점에서 디스패처를 명시적으로 지정하는 것보다 바깥 스코프의 디스패처를 사용하는 것이 좋은 관행으로 여겨져요.
loadContributorsConcurrent()를Dispatchers.Default를 인자로 넘기지 않고 정의하면 이 함수를Default디스패처, main UI 스레드, 또는 커스텀 디스패처 등 어떤 컨텍스트에서도 호출할 수 있어요. - 나중에 보겠지만 테스트에서
loadContributorsConcurrent()를 호출할 때TestDispatcher가 있는 컨텍스트에서 호출할 수 있어서 테스트가 단순해져요. 그래서 이 해결책이 훨씬 유연하죠. - 디스패처를 호출하는 쪽에서 지정하려면
loadContributorsConcurrent가 상속된 컨텍스트에서 코루틴을 시작하게 두면서 프로젝트에 다음 변경을 적용해요.
launch(Dispatchers.Default) {
val users = loadContributorsConcurrent(service, req)
withContext(Dispatchers.Main) {
updateResults(users, startTime)
}
}
updateResults()는 main UI 스레드에서 호출돼야 하므로Dispatchers.Main의 컨텍스트로 호출해요.withContext()는 지정된 코루틴 컨텍스트로 주어진 코드를 호출하고, 완료될 때까지 중단된 상태로 있으며 결과를 반환해요. 이를 표현하는 대안이지만 더 장황한 방법은 새 코루틴을 시작하고 명시적으로 완료를 기다리는(중단하는) 것, 즉launch(context) { ... }.join()이에요.- 코드를 실행하고 코루틴이 스레드 풀의 스레드에서 실행되는지 확인해요.
구조적 동시성
- 코루틴 스코프는 서로 다른 코루틴 사이의 구조와 부모-자식 관계를 담당해요. 새 코루틴은 보통 스코프 안에서 시작돼야 해요.
- 코루틴 컨텍스트는 주어진 코루틴을 실행하는 데 사용되는 추가 기술 정보를 저장해요. 예를 들어 코루틴 커스텀 이름이나 코루틴이 예약될 스레드를 지정하는 디스패처 같은 것들이죠.
launch, async, runBlocking으로 새 코루틴을 시작하면 해당 스코프를 자동으로 만들어요. 이 함수들은 모두 리시버가 있는 람다를 인자로 받고, CoroutineScope가 암시적 리시버 타입이에요.
launch { /* this: CoroutineScope */ }
- 새 코루틴은 스코프 안에서만 시작할 수 있어요.
launch와async는CoroutineScope의 확장으로 선언되므로 호출할 때 항상 암시적 또는 명시적 리시버를 전달해야 해요.runBlocking이 시작한 코루틴은 유일한 예외인데,runBlocking이 최상위 함수로 정의되기 때문이에요. 하지만 현재 스레드를 막기 때문에 주로main()함수와 테스트에서 브리지 함수로 사용되도록 만들어졌어요.
runBlocking, launch, async 안의 새 코루틴은 스코프 안에서 자동으로 시작돼요.
import kotlinx.coroutines.*
fun main() = runBlocking { /* this: CoroutineScope */
launch { /* ... */ }
// 다음과 같음:
this.launch { /* ... */ }
}
runBlocking 안에서 launch를 호출하면 CoroutineScope 타입의 암시적 리시버에 대한 확장으로 호출돼요. 또는 명시적으로 this.launch라고 쓸 수도 있어요.
중첩 코루틴(이 예시에서 launch가 시작한 것)은 바깥 코루틴(runBlocking이 시작한 것)의 자식으로 간주될 수 있어요. 이 "부모-자식" 관계는 스코프를 통해 동작해요. 자식 코루틴은 부모 코루틴에 해당하는 스코프에서 시작되죠.
새 코루틴을 시작하지 않고도 coroutineScope 함수를 사용해 새 스코프를 만들 수 있어요. 바깥 스코프에 접근할 수 없는 suspend 함수 안에서 새 코루틴을 구조적으로 시작하려면, 이 suspend 함수가 호출된 바깥 스코프의 자식이 자동으로 되는 새 코루틴 스코프를 만들 수 있어요. loadContributorsConcurrent()가 좋은 예시예요.
GlobalScope.async나 GlobalScope.launch를 사용해 전역 스코프에서 새 코루틴을 시작할 수도 있어요. 이렇게 하면 최상위 "독립적인" 코루틴이 생성돼요.
코루틴 구조의 이면에 있는 메커니즘을 **구조적 동시성(structured concurrency)**이라고 불러요. 전역 스코프보다 다음과 같은 이점을 제공해요.
- 스코프는 일반적으로 자식 코루틴을 책임지며, 자식 코루틴의 수명은 스코프의 수명에 연결돼요.
- 스코프는 문제가 생기거나 사용자가 마음을 바꿔 연산을 취소하기로 하면 자식 코루틴을 자동으로 취소할 수 있어요.
- 스코프는 모든 자식 코루틴의 완료를 자동으로 기다려요. 따라서 스코프가 코루틴에 해당하면 부모 코루틴은 그 스코프에서 시작된 모든 코루틴이 완료될 때까지 완료되지 않아요.
GlobalScope.async를 사용하면 여러 코루틴을 더 작은 스코프에 묶는 구조가 없어요. 전역 스코프에서 시작된 코루틴은 모두 독립적이며, 그 수명은 전체 애플리케이션의 수명에 의해서만 제한돼요. 전역 스코프에서 시작된 코루틴에 대한 참조를 저장하고 완료를 기다리거나 명시적으로 취소할 수는 있지만, 구조적 동시성처럼 자동으로 일어나지는 않아요.
기여자 로딩 취소
기여자 목록을 로드하는 함수의 두 가지 버전을 만들어요. 부모 코루틴을 취소하려 할 때 두 버전이 어떻게 동작하는지 비교해 봐요. 첫 번째 버전은 coroutineScope를 사용해 모든 자식 코루틴을 시작하고, 두 번째는 GlobalScope를 사용해요.
Request5Concurrent.kt에서loadContributorsConcurrent()함수에 3초 지연을 추가해요.
suspend fun loadContributorsConcurrent(
service: GitHubService,
req: RequestData
): List<User> = coroutineScope {
// ...
async {
log("starting loading for ${repo.name}")
delay(3000)
// repo 기여자 로드
}
// ...
}
이 지연은 요청을 보내는 모든 코루틴에 영향을 줘서, 코루틴이 시작된 후 요청이 전송되기 전에 로딩을 취소할 충분한 시간을 확보해요.
- 로딩 함수의 두 번째 버전을 만들어요.
loadContributorsConcurrent()의 구현을Request5NotCancellable.kt의loadContributorsNotCancellable()에 복사하고 새coroutineScope생성을 제거해요. - 이제
async호출이 해석되지 않으므로GlobalScope.async를 사용해 시작해요.
suspend fun loadContributorsNotCancellable(
service: GitHubService,
req: RequestData
): List<User> { // #1
// ...
GlobalScope.async { // #2
log("starting loading for ${repo.name}")
// repo 기여자 로드
}
// ...
return deferreds.awaitAll().flatten().aggregate() // #3
}
- 이제 함수는 람다 안의 마지막 표현식이 아니라 결과를 직접 반환해요(
#1과#3줄). - 모든 "contributors" 코루틴은 코루틴 스코프의 자식이 아니라
GlobalScope안에서 시작돼요(#2줄). - 프로그램을 실행하고 CONCURRENT 옵션을 선택해 기여자를 로드해요.
- 모든 "contributors" 코루틴이 시작될 때까지 기다린 다음 Cancel을 클릭해요. 로그에 새 결과가 없는데, 이는 모든 요청이 실제로 취소됐다는 뜻이에요.
2896 [AWT-EventQueue-0 @coroutine#1] INFO Contributors - kotlin: loaded 40 repos
2901 [DefaultDispatcher-worker-2 @coroutine#4] INFO Contributors - starting loading for kotlin-koans
...
2909 [DefaultDispatcher-worker-5 @coroutine#36] INFO Contributors - starting loading for mpp-example
/* 'cancel' 클릭 */
/* 요청이 전송되지 않음 */
- 5단계를 반복하되, 이번에는
NOT_CANCELLABLE옵션을 선택해요.
2570 [AWT-EventQueue-0 @coroutine#1] INFO Contributors - kotlin: loaded 30 repos
2579 [DefaultDispatcher-worker-1 @coroutine#4] INFO Contributors - starting loading for kotlin-koans
...
2586 [DefaultDispatcher-worker-6 @coroutine#36] INFO Contributors - starting loading for mpp-example
/* 'cancel' 클릭 */
/* 하지만 모든 요청은 여전히 전송됨: */
6402 [DefaultDispatcher-worker-5 @coroutine#4] INFO Contributors - kotlin-koans: loaded 45 contributors
...
9555 [DefaultDispatcher-worker-8 @coroutine#36] INFO Contributors - mpp-example: loaded 8 contributors
이 경우 어떤 코루틴도 취소되지 않고 모든 요청이 여전히 전송돼요.
- "contributors" 프로그램에서 취소가 어떻게 트리거되는지 확인해요. Cancel 버튼을 클릭하면 main "loading" 코루틴이 명시적으로 취소되고 자식 코루틴이 자동으로 취소돼요.
interface Contributors {
fun loadContributors() {
// ...
when (getSelectedVariant()) {
CONCURRENT -> {
launch {
val users = loadContributorsConcurrent(service, req)
updateResults(users, startTime)
}.setUpCancellation() // #1
}
}
}
private fun Job.setUpCancellation() {
val loadingJob = this // #2
// 'cancel' 버튼이 클릭되면 loading job을 취소해요:
val listener = ActionListener {
loadingJob.cancel() // #3
updateLoadingStatus(CANCELED)
}
// 'cancel' 버튼에 리스너를 추가해요:
addCancelListener(listener)
// loading job이 완료된 후
// 상태를 갱신하고 리스너를 제거해요
}
}
launch 함수는 Job 인스턴스를 반환해요. Job은 모든 데이터를 로드하고 결과를 갱신하는 "loading coroutine"에 대한 참조를 저장해요. 그 위에 setUpCancellation() 확장 함수를 호출할 수 있는데(#1 줄), Job 인스턴스를 리시버로 전달해요.
이를 표현하는 또 다른 방법은 명시적으로 다음과 같이 쓰는 거예요.
val job = launch { }
job.setUpCancellation()
- 가독성을 위해 함수 안에서
setUpCancellation()함수의 리시버를 새loadingJob변수로 참조할 수 있어요(#2줄). - 그런 다음 Cancel 버튼에 리스너를 추가해 클릭하면
loadingJob이 취소되게 할 수 있어요(#3줄).
구조적 동시성에서는 부모 코루틴만 취소하면 되고, 이 취소가 자동으로 모든 자식 코루틴에 전파돼요.
바깥 스코프의 컨텍스트 사용
주어진 스코프 안에서 새 코루틴을 시작할 때, 모두가 같은 컨텍스트로 실행되도록 하는 것이 훨씬 쉬워요. 필요할 때 컨텍스트를 교체하는 것도 훨씬 쉬워요.
이제 바깥 스코프의 디스패처를 사용하는 방식이 어떻게 동작하는지 배워 봐요. coroutineScope나 코루틴 빌더가 만든 새 스코프는 항상 바깥 스코프의 컨텍스트를 상속해요. 이 경우 바깥 스코프는 suspend loadContributorsConcurrent() 함수가 호출된 스코프예요.
launch(Dispatchers.Default) { // outer scope
val users = loadContributorsConcurrent(service, req)
// ...
}
모든 중첩 코루틴은 상속된 컨텍스트로 자동 시작돼요. 디스패처는 이 컨텍스트의 일부예요. 그래서 async가 시작한 모든 코루틴은 default 디스패처의 컨텍스트로 시작돼요.
suspend fun loadContributorsConcurrent(
service: GitHubService, req: RequestData
): List<User> = coroutineScope {
// this scope는 바깥 스코프의 컨텍스트를 상속해요
// ...
async { // 상속된 컨텍스트로 시작된 중첩 코루틴
// ...
}
// ...
}
구조적 동시성에서는 주요 컨텍스트 요소(디스패처 같은)를 최상위 코루틴을 만들 때 한 번만 지정할 수 있어요. 그러면 모든 중첩 코루틴이 컨텍스트를 상속하고 필요할 때만 수정해요.
Android 같은 UI 애플리케이션용으로 코루틴으로 코드를 작성할 때, 상위 코루틴에는 기본적으로
CoroutineDispatchers.Main을 사용하고 다른 스레드에서 코드를 실행해야 할 때 명시적으로 다른 디스패처를 넣는 것이 일반적인 관행이에요.
진행 상태 표시
일부 저장소의 정보는 꽤 빨리 로드되지만, 사용자는 모든 데이터가 로드된 후에만 결과 목록을 볼 수 있어요. 그때까지 로더 아이콘이 진행 상태를 보여주며 돌지만, 현재 상태나 이미 로드된 기여자에 대한 정보는 없어요.
각 저장소의 데이터를 로드한 후 중간 결과를 더 일찍 보여주고 모든 기여자를 표시할 수 있어요.
이 기능을 구현하려면 src/tasks/Request6Progress.kt에서 UI를 갱신하는 로직을 콜백으로 전달해서 각 중간 상태에서 호출되게 해야 해요.
suspend fun loadContributorsProgress(
service: GitHubService,
req: RequestData,
updateResults: suspend (List<User>, completed: Boolean) -> Unit
) {
// 데이터 로드
// 중간 상태에서 `updateResults()` 호출
}
Contributors.kt의 호출 지점에서 PROGRESS 옵션에 대해 Main 스레드에서 결과를 갱신하도록 콜백이 전달돼요.
launch(Dispatchers.Default) {
loadContributorsProgress(service, req) { users, completed ->
withContext(Dispatchers.Main) {
updateResults(users, startTime, completed)
}
}
}
updateResults()매개변수는loadContributorsProgress()에서suspend로 선언돼요. 해당 람다 인자 안에서suspend함수인withContext를 호출하기 위해 필요해요.updateResults()콜백은 로딩이 완료됐고 결과가 최종인지 지정하는 추가 Boolean 매개변수를 인자로 받아요.
태스크 6
Request6Progress.kt 파일에서 중간 진행 상태를 보여주는 loadContributorsProgress() 함수를 구현해요. Request4Suspend.kt의 loadContributorsSuspend() 함수에 기반해요.
- 동시성이 없는 간단한 버전을 사용해요. 다음 섹션에서 나중에 추가할 거예요.
- 중간 기여자 목록은 각 저장소에 대해 로드된 사용자 목록 그대로가 아니라 "집계된(aggregated)" 상태로 표시돼야 해요.
- 각 새 저장소의 데이터가 로드될 때 각 사용자의 총 기여 횟수가 증가해야 해요.
태스크 6의 해결 방법
로드된 기여자의 중간 목록을 "집계된" 상태로 저장하려면 사용자 목록을 저장하는 allUsers 변수를 정의하고, 각 새 저장소의 기여자가 로드된 후 그 값을 갱신해요.
suspend fun loadContributorsProgress(
service: GitHubService,
req: RequestData,
updateResults: suspend (List<User>, completed: Boolean) -> Unit
) {
val repos = service
.getOrgRepos(req.org)
.also { logRepos(req, it) }
.bodyList()
var allUsers = emptyList<User>()
for ((index, repo) in repos.withIndex()) {
val users = service.getRepoContributors(req.org, repo.name)
.also { logUsers(repo, it) }
.bodyList()
allUsers = (allUsers + users).aggregate()
updateResults(allUsers, index == repos.lastIndex)
}
}
연속 vs 동시
updateResults() 콜백은 각 요청이 완료된 후 호출돼요.
이 코드는 동시성을 포함하지 않아요. 순차적이라 동기화가 필요 없어요.
가장 좋은 방법은 요청을 동시에 보내고 각 저장소의 응답을 받은 후 중간 결과를 갱신하는 거예요.
동시성을 추가하려면 채널을 사용해요.
채널
공유된 가변 상태로 코드를 작성하는 것은 꽤 어렵고 오류가 발생하기 쉬워요(콜백을 사용한 해결책처럼). 더 간단한 방법은 공통의 가변 상태를 사용하는 대신 통신으로 정보를 공유하는 거예요. 코루틴은 채널을 통해 서로 통신할 수 있어요.
채널은 코루틴 사이에 데이터를 전달할 수 있게 하는 커뮤니케이션 프리미티브예요. 한 코루틴은 채널에 일부 정보를 보낼 수 있고, 다른 코루틴은 그 정보를 채널에서 받을 수 있어요.
정보를 보내는(생산하는) 코루틴을 종종 **프로듀서(producer)**라고 하고, 정보를 받는(소비하는) 코루틴을 **컨슈머(consumer)**라고 해요. 하나 또는 여러 코루틴이 같은 채널에 정보를 보낼 수 있고, 하나 또는 여러 코루틴이 채널에서 데이터를 받을 수 있어요.
많은 코루틴이 같은 채널에서 정보를 받을 때 각 요소는 컨슈머 중 하나가 한 번만 처리해요. 요소가 처리되면 즉시 채널에서 제거돼요.
채널을 요소 모음, 더 정확히는 한쪽 끝에 요소가 추가되고 다른 쪽에서 받는 **큐(queue)**와 비슷하다고 생각할 수 있어요. 하지만 중요한 차이가 있어요. 컬렉션과 달리, 동기화된 버전이라도 채널은 send()와 receive() 연산을 중단할 수 있어요. 채널이 비어 있거나 가득 차 있을 때 그렇게 되죠. 채널 크기에 상한이 있으면 채널이 가득 찰 수 있어요.
Channel은 SendChannel, ReceiveChannel, Channel이라는 세 가지 서로 다른 인터페이스로 표현되며, 마지막 것이 처음 두 개를 확장해요. 보통 채널을 만들어 프로듀서에게 SendChannel 인스턴스로 주면 프로듀서만 채널에 정보를 보낼 수 있어요. 컨슈머에게는 ReceiveChannel 인스턴스로 채널을 줘서 컨슈머만 채널에서 받게 해요. send와 receive 메서드는 모두 suspend로 선언돼요.
interface SendChannel<in E> {
suspend fun send(element: E)
fun close(): Boolean
}
interface ReceiveChannel<out E> {
suspend fun receive(): E
}
interface Channel<E> : SendChannel<E>, ReceiveChannel<E>
프로듀서는 더 이상 요소가 오지 않음을 나타내기 위해 채널을 닫을 수 있어요.
라이브러리에는 여러 유형의 채널이 정의돼 있어요. 내부에 저장할 수 있는 요소 수와 send() 호출이 중단될 수 있는지 여부가 서로 달라요. 모든 채널 유형에서 receive() 호출은 비슷하게 동작해요. 채널이 비어 있지 않으면 요소를 받고, 그렇지 않으면 중단돼요.
Unlimited 채널 — 무제한 채널은 큐에 가장 가까운 유사체예요. 프로듀서는 이 채널에 요소를 보낼 수 있고 무한정 계속 커져요. send() 호출은 절대 중단되지 않아요. 프로그램의 메모리가 부족하면 OutOfMemoryException을 얻게 돼요. 무제한 채널과 큐의 차이는 컨슈머가 빈 채널에서 받으려 하면 새 요소가 보내질 때까지 중단된다는 점이에요.
Buffered 채널 — 버퍼 채널의 크기는 지정된 숫자로 제한돼요. 프로듀서는 크기 제한에 도달할 때까지 이 채널에 요소를 보낼 수 있어요. 모든 요소는 내부에 저장돼요. 채널이 가득 차면 다음 send() 호출은 더 많은 여유 공간이 생길 때까지 중단돼요.
Rendezvous 채널 — "Rendezvous" 채널은 버퍼가 없는 채널로, 크기가 0인 버퍼 채널과 같아요. send() 또는 receive() 함수 중 하나는 다른 하나가 호출될 때까지 항상 중단돼요.
send() 함수가 호출되고 요소를 처리할 준비가 된 중단된 receive() 호출이 없으면 send()는 중단돼요. 마찬가지로 receive() 함수가 호출되고 채널이 비어 있거나, 다시 말해 요소를 보낼 준비가 된 중단된 send() 호출이 없으면 receive() 호출은 중단돼요.
"rendezvous"(동의된 시간과 장소에서의 만남)라는 이름은 send()와 receive()가 "시간에 맞춰 만나야" 한다는 사실을 가리켜요.
Conflated 채널 — conflated 채널에 보내진 새 요소는 이전에 보낸 요소를 덮어써서, 수신자는 항상 최신 요소만 받아요. send() 호출은 절대 중단되지 않아요.
채널을 만들 때 유형이나 버퍼 크기를 지정해요(버퍼 채널이 필요하면).
val rendezvousChannel = Channel<String>()
val bufferedChannel = Channel<String>(10)
val conflatedChannel = Channel<String>(CONFLATED)
val unlimitedChannel = Channel<String>(UNLIMITED)
기본적으로 "Rendezvous" 채널이 생성돼요.
다음 태스크에서는 "Rendezvous" 채널, 두 개의 프로듀서 코루틴, 하나의 컨슈머 코루틴을 만들 거예요.
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.*
fun main() = runBlocking<Unit> {
val channel = Channel<String>()
launch {
channel.send("A1")
channel.send("A2")
log("A done")
}
launch {
channel.send("B1")
log("B done")
}
launch {
repeat(3) {
val x = channel.receive()
log(x)
}
}
}
fun log(message: Any?) {
println("[${Thread.currentThread().name}] $message")
}
채널을 더 잘 이해하려면 이 비디오를 시청해 보세요.
태스크 7
src/tasks/Request7Channels.kt에서 모든 GitHub 기여자를 동시에 요청하고 동시에 중간 진행 상태를 보여주는 loadContributorsChannels() 함수를 구현해요.
Request5Concurrent.kt의 loadContributorsConcurrent()와 Request6Progress.kt의 loadContributorsProgress() 이전 함수들을 사용해요.
태스크 7의 힌트
서로 다른 저장소의 기여자 목록을 동시에 받는 서로 다른 코루틴이 받은 모든 결과를 같은 채널로 보낼 수 있어요.
val channel = Channel<List<User>>()
for (repo in repos) {
launch {
val users = TODO()
// ...
channel.send(users)
}
}
그런 다음 이 채널의 요소를 하나씩 받아 처리할 수 있어요.
repeat(repos.size) {
val users = channel.receive()
// ...
}
receive() 호출은 순차적이므로 추가 동기화가 필요 없어요.
태스크 7의 해결 방법
loadContributorsProgress() 함수처럼 "all contributors" 목록의 중간 상태를 저장할 allUsers 변수를 만들 수 있어요. 채널에서 받은 각 새 목록이 모든 사용자 목록에 추가돼요. 결과를 집계하고 updateResults 콜백으로 상태를 갱신해요.
suspend fun loadContributorsChannels(
service: GitHubService,
req: RequestData,
updateResults: suspend (List<User>, completed: Boolean) -> Unit
) = coroutineScope {
val repos = service
.getOrgRepos(req.org)
.also { logRepos(req, it) }
.bodyList()
val channel = Channel<List<User>>()
for (repo in repos) {
launch {
val users = service.getRepoContributors(req.org, repo.name)
.also { logUsers(repo, it) }
.bodyList()
channel.send(users)
}
}
var allUsers = emptyList<User>()
repeat(repos.size) {
val users = channel.receive()
allUsers = (allUsers + users).aggregate()
updateResults(allUsers, it == repos.lastIndex)
}
}
- 서로 다른 저장소의 결과는 준비되는 즉시 채널에 추가돼요. 처음에는 모든 요청이 전송되고 데이터가 수신되지 않았을 때
receive()호출이 중단돼요. 이 경우 전체 "load contributors" 코루틴이 중단돼요. - 그런 다음 사용자 목록이 채널로 보내지면 "load contributors" 코루틴이 재개되고,
receive()호출이 이 목록을 반환하며 결과가 즉시 갱신돼요.
이제 프로그램을 실행하고 CHANNELS 옵션을 선택해 기여자를 로드하고 결과를 확인할 수 있어요.
코루틴이나 채널 모두 동시성과 함께 오는 복잡성을 완전히 없애지는 못하지만, 무슨 일이 일어나고 있는지 이해해야 할 때 삶을 더 쉽게 만들어요.
코루틴 테스트
이제 모든 해결책이 동시 코루틴 해결책이 suspend 함수 해결책보다 빠른지, 그리고 채널 해결책이 단순한 "progress" 해결책보다 빠른지 확인하기 위해 모두 테스트해 봐요.
다음 태스크에서는 해결책의 총 실행 시간을 비교할 거예요. GitHub 서비스를 모의(mock)하고 이 서비스가 주어진 타임아웃 후에 결과를 반환하게 만들 거예요.
repos request - 1000 ms 지연 후 응답 반환
repo-1 - 1000 ms 지연
repo-2 - 1200 ms 지연
repo-3 - 800 ms 지연
suspend 함수가 있는 순차 해결책은 약 4000 ms(4000 = 1000 + (1000 + 1200 + 800))가 걸려야 해요. 동시 해결책은 약 2200 ms(2200 = 1000 + max(1000, 1200, 800))가 걸려야 해요.
진행 상태를 보여주는 해결책에서는 타임스탬프로 중간 결과도 확인할 수 있어요.
해당 테스트 데이터는 test/contributors/testData.kt에 정의돼 있고, Request4SuspendKtTest, Request7ChannelsKtTest 등의 파일에는 mock 서비스 호출을 사용하는 간단한 테스트가 들어 있어요.
하지만 여기에는 두 가지 문제가 있어요.
- 이 테스트들은 실행하는 데 너무 오래 걸려요. 각 테스트는 약 2~4초가 걸리고, 매번 결과를 기다려야 해요. 그다지 효율적이지 않죠.
- 해결책이 실행되는 정확한 시간에 의존할 수 없어요. 코드를 준비하고 실행하는 데 추가 시간이 여전히 들기 때문이에요. 상수를 추가할 수 있지만 그러면 머신마다 시간이 달라져요. mock 서비스 지연은 이 상수보다 커야 그 차이를 볼 수 있어요. 상수가 0.5초라면 지연을 0.1초로 만드는 것으로는 충분하지 않아요.
더 좋은 방법은 같은 코드를 여러 번 실행하면서 시간을 테스트하는 특수 프레임워크를 사용하는 것(총 시간이 더 늘어나지만)인데, 배우고 설정하기가 복잡해요.
이 문제들을 해결하고 제공된 테스트 지연으로 해결책이 기대대로, 하나가 다른 것보다 빠르게, 동작하도록 하려면 특수 테스트 디스패처로 **가상 시간(virtual time)**을 사용해요. 이 디스패처는 시작부터 지난 가상 시간을 추적하고 실제 시간으로는 모든 것을 즉시 실행해요. 이 디스패처에서 코루틴을 실행하면 delay가 즉시 반환되고 가상 시간이 앞으로 진행돼요.
이 메커니즘을 사용하는 테스트는 빠르게 실행되지만, 가상 시간의 서로 다른 순간에 무슨 일이 일어나는지는 여전히 확인할 수 있어요. 총 실행 시간이 크게 줄어들어요.
가상 시간을 사용하려면 runBlocking 호출을 runTest로 바꿔요. runTest는 TestScope에 대한 확장 람다를 인자로 받아요. 이 특수 스코프 안의 suspend 함수에서 delay를 호출하면 delay가 실제 시간으로 지연하는 대신 가상 시간을 늘려요.
@Test
fun testDelayInSuspend() = runTest {
val realStartTime = System.currentTimeMillis()
val virtualStartTime = currentTime
foo()
println("${System.currentTimeMillis() - realStartTime} ms") // ~ 6 ms
println("${currentTime - virtualStartTime} ms") // 1000 ms
}
suspend fun foo() {
delay(1000) // 지연 없이 자동 진행
println("foo") // foo()가 호출되면 즉시 실행
}
TestScope의 currentTime 속성으로 현재 가상 시간을 확인할 수 있어요.
이 예시의 실제 실행 시간은 몇 밀리초인 반면, 가상 시간은 지연 인자와 같은 1000밀리초예요.
자식 코루틴에서 "가상" delay의 전체 효과를 얻으려면 모든 자식 코루틴을 TestDispatcher로 시작해요. 그렇지 않으면 동작하지 않아요. 다른 디스패처를 제공하지 않으면 이 디스패처가 다른 TestScope에서 자동으로 상속돼요.
@Test
fun testDelayInLaunch() = runTest {
val realStartTime = System.currentTimeMillis()
val virtualStartTime = currentTime
bar()
println("${System.currentTimeMillis() - realStartTime} ms") // ~ 11 ms
println("${currentTime - virtualStartTime} ms") // 1000 ms
}
suspend fun bar() = coroutineScope {
launch {
delay(1000) // 지연 없이 자동 진행
println("bar") // bar()가 호출되면 즉시 실행
}
}
위 예시에서 launch가 Dispatchers.Default의 컨텍스트로 호출되면 테스트가 실패해요. 작업이 아직 완료되지 않았다는 예외를 얻게 되죠.
loadContributorsConcurrent() 함수는 Dispatchers.Default 디스패처로 수정하지 않고 상속된 컨텍스트로 자식 코루틴을 시작할 때만 이런 방식으로 테스트할 수 있어요.
디스패처 같은 컨텍스트 요소를 함수를 정의할 때가 아니라 호출할 때 지정할 수 있어서 더 큰 유연성과 더 쉬운 테스트가 가능해요.
가상 시간을 지원하는 테스트 API는 Experimental이며 향후 변경될 수 있어요.
기본적으로 컴파일러는 experimental 테스트 API를 사용하면 경고를 보여줘요. 이 경고를 억제하려면 테스트 함수나 테스트를 포함하는 전체 클래스에 @OptIn(ExperimentalCoroutinesApi::class)를 붙여요. experimental API를 사용하고 있음을 컴파일러에 지시하는 컴파일러 인자를 추가해요.
compileTestKotlin {
kotlinOptions {
freeCompilerArgs += "-Xuse-experimental=kotlin.Experimental"
}
}
이 튜토리얼에 해당하는 프로젝트에서는 컴파일러 인자가 이미 Gradle 스크립트에 추가돼 있어요.
태스크 8
tests/tasks/의 다음 테스트들을 실제 시간 대신 가상 시간을 사용하도록 리팩터링해요.
Request4SuspendKtTest.ktRequest5ConcurrentKtTest.ktRequest6ProgressKtTest.ktRequest7ChannelsKtTest.kt
리팩터링 전후의 총 실행 시간을 비교해요.
태스크 8의 힌트
runBlocking호출을runTest로,System.currentTimeMillis()를currentTime으로 바꿔요.
@Test
fun test() = runTest {
val startTime = currentTime
// 동작
val totalTime = currentTime - startTime
// 결과 테스트
}
- 정확한 가상 시간을 확인하는 단언(assertions)의 주석을 해제해요.
@UseExperimental(ExperimentalCoroutinesApi::class)를 추가하는 것을 잊지 마세요.
태스크 8의 해결 방법
다음은 동시와 채널 사례의 해결책이에요.
fun testConcurrent() = runTest {
val startTime = currentTime
val result = loadContributorsConcurrent(MockGithubService, testRequestData)
Assert.assertEquals("Wrong result for 'loadContributorsConcurrent'", expectedConcurrentResults.users, result)
val totalTime = currentTime - startTime
Assert.assertEquals(
"The calls run concurrently, so the total virtual time should be 2200 ms: " +
"1000 for repos request plus max(1000, 1200, 800) = 1200 for concurrent contributors requests)",
expectedConcurrentResults.timeFromStart, totalTime
)
}
먼저 결과가 정확히 예상 가상 시간에 사용 가능한지 확인한 다음 결과 자체를 확인해요.
fun testChannels() = runTest {
val startTime = currentTime
var index = 0
loadContributorsChannels(MockGithubService, testRequestData) { users, _ ->
val expected = concurrentProgressResults[index++]
val time = currentTime - startTime
Assert.assertEquals(
"Expected intermediate results after ${expected.timeFromStart} ms:",
expected.timeFromStart, time
)
Assert.assertEquals("Wrong intermediate results after $time:", expected.users, users)
}
}
채널이 있는 마지막 버전의 첫 번째 중간 결과는 progress 버전보다 더 일찍 사용 가능해지고, 가상 시간을 사용하는 테스트에서 그 차이를 볼 수 있어요.
나머지 "suspend"와 "progress" 태스크의 테스트는 매우 비슷해요. 프로젝트의
solutions브랜치에서 찾을 수 있어요.
다음 단계
- KotlinConf의 Asynchronous Programming with Kotlin 워크숍을 확인해 보세요.
- 가상 시간과 experimental 테스트 패키지 사용에 대해 더 알아보세요.