동시성처리와 병렬성처리는 유사해보이지만 큰 차이가있으며 간략하게 간략하게 설명될수 있습니다.
- 동시성 : 한사람이 커피주문 받고 만들기를 동시에한다. ( 커피머신을 이용하면서도 주문을 받음)
- 병렬성 : 커피주문을 받는사람과 만드는 사람이 각각 따로 있다.
- 순차성 : 한사람이 커피주문을 받고 커피를 만들지만~ 커피가 완성될때까지 다음손님 주문을 받지 못한다.
🧩 1. 전체 테스트 구조
이 파일은 코루틴, 리액터(Reactor), CompletableFuture, Java Stream, Akka Streams, 액터(Actor) 등을 사용해서
"비동기 동시성" 그리고 "액터(Actor) 기반 메시지 통신" 테스트를 다양한 방법으로 검증하는 것입니다.
구체적으로는:
| 구분 | 내용 |
|---|---|
| 기초 비동기 처리 | Coroutine, Mono, CompletableFuture, Java Stream으로 "Hello World"를 만드는 간단한 동시성 테스트 |
| 병렬 실행 테스트 | 3개의 작업(문자열 생성+지연 메소드 호출 2개)을 병렬로 수행하여 하나의 문자열로 합치는 테스트 |
| Akka Stream 테스트 | Akka Streams를 사용하여 비동기 스트림 병합 테스트 |
| Actor Ask 패턴 테스트 | HelloActor를 대상으로 ask (질문→응답받기) 패턴 테스트 |
| Kotlin 기반 커스텀 KHelloActor 테스트 | 별도로 작성한 코루틴 기반 액터 (KHelloActor) 테스트 |
🧩 2. 주요 테스트 코드 설명
(1) 기본 비동기 처리 (Hello → Hello World)
| 방법 | 요약 |
|---|---|
Coroutine | async로 비동기 작업하고 await()로 결과 가져옴 |
Mono | 리액터 기반 Mono 스트림 처리 |
CompletableFuture | 자바 기본 비동기 API 활용 |
Java Stream | 스트림 map 연산으로 변환 처리 후 CompletableFuture |
공통 목표: "Hello World" 문자열 생성 및 검증
testHelloByCoroutine
@Test
fun testHelloByCoroutine() = runBlocking {
// Test using CompletableFuture to concatenate strings asynchronously
val input = "Hello"
val result = withContext(Dispatchers.Default) {
val part1 = async { input + " World" }
part1.await()
}
assertEquals("Hello World", result)
}
testHelloByMono
@Test
fun testHelloByMono() {
// Test using Reactor Mono to concatenate strings asynchronously
val input = "Hello"
val monoResult = Mono.just(input)
.map { it + " World" }
.block()
assertEquals("Hello World", monoResult)
}
testHelloByCompletableFuture
@Test
fun testHelloByCompletableFuture() {
// Test using CompletableFuture to concatenate strings asynchronously
val future = CompletableFuture.supplyAsync { "Hello" }
.thenApplyAsync { result -> result + " World" }
val result = future.get()
assertEquals("Hello World", result)
}
testHelloByJavaStream
@Test
fun testHelloByJavaStream() {
// Test using CompletableFuture with Java Streams to concatenate strings asynchronously
val future = CompletableFuture.supplyAsync {
listOf("Hello")
.stream()
.map { it + " World" }
.findFirst()
.orElse("")
}
val result = future.get()
assertEquals("Hello World", result)
}
(2) 동시성 처리 (Concurrency)
getDelayString()와 getDelayString2()는 각각 600ms, 500ms 지연 후 "SlowWorld1", "SlowWorld2"를 반환합니다.
이걸 병렬로 돌려서:
Hello World SlowWorld1 SlowWorld2
라는 최종 문자열을 만들고 검증합니다.
방법별 특징:
Coroutine
async를 3개 띄워await로 병합Mono
zipWith으로 여러 Mono를 병렬 조합Flux
flatMap으로 병렬 작업 후 결과 모음CompletableFuture
thenCombineAsync로 결과 합성Java Stream
Stream 내부에서 Futurejoin()해서 결과 병합Akka Stream
Source를 연결(zipWith)해서 결과를 스트림 합성
testConcurrentByCoroutine
@Test
fun testConcurrentByCoroutine() = runBlocking {
val input = "Hello"
val result = withContext(Dispatchers.Default) {
val part1 = async { input + " World" }
val part2 = async { getDelayString() }
val part3 = async { getDelayString2() }
"${part1.await()} ${part2.await()} ${part3.await()}"
}
assertEquals("Hello World SlowWorld1 SlowWorld2", result)
}
testConcurrentByMono
@Test
fun testConcurrentByMono() {
val input = "Hello"
val monoResult = Mono.just(input)
.map { it + " World" }
.zipWith(Mono.fromCallable { getDelayString() })
.zipWith(Mono.fromCallable { getDelayString2() }) { part1, part2 ->
"${part1.t1} ${part1.t2} $part2"
}
.block()
assertEquals("Hello World SlowWorld1 SlowWorld2", monoResult)
}
testConcurrentByFlux
@Test
fun testConcurrentByFlux() {
val input = listOf("Hello", "World")
val fluxResult = Flux.fromIterable(input)
.flatMap { data ->
when (data) {
"Hello" -> Mono.fromCallable { getDelayString() }
"World" -> Mono.fromCallable { getDelayString2() }
else -> Mono.empty()
}
}
.collectList()
.map { results -> results.joinToString(" ") }
.block()
assertEquals("SlowWorld1 SlowWorld2", fluxResult)
}
testConcurrentByCompletableFuture
@Test
fun testConcurrentByCompletableFuture() {
val future = CompletableFuture.supplyAsync { "Hello" }
.thenApplyAsync { result -> result + " World" }
.thenCombineAsync(CompletableFuture.supplyAsync { getDelayString() }) { part1, part2 ->
"$part1 $part2"
}
.thenCombineAsync(CompletableFuture.supplyAsync { getDelayString2() }) { part1, part3 ->
"$part1 $part3"
}
val result = future.get()
assertEquals("Hello World SlowWorld1 SlowWorld2", result)
}
testConcurrentByJavsStream
@Test
fun testConcurrentByJavsStream() {
val result = listOf("Hello")
.stream()
.map { input -> input + " World" }
.map { part1 ->
val part2 = CompletableFuture.supplyAsync { getDelayString() }.join()
val part3 = CompletableFuture.supplyAsync { getDelayString2() }.join()
"$part1 $part2 $part3"
}
.findFirst()
.orElse("")
assertEquals("Hello World SlowWorld1 SlowWorld2", result)
}
testConcurrentByAkkaStream
@Test
fun testConcurrentByAkkaStream() {
val materializer = Materializer.createMaterializer(actorSystem.system())
val source = Source.single("Hello")
.map { it + " World" }
.zipWith(Source.single(getDelayString())) { part1, part2 -> Pair(part1, part2) }
.zipWith(Source.single(getDelayString2())) { pair, part3 ->
"${pair.first} ${pair.second} $part3"
}
.runWith(Sink.head(), materializer)
val result = source.toCompletableFuture().get()
assertEquals("Hello World SlowWorld1 SlowWorld2", result)
}
(3) 액터(Actor) 기반 메시지 송수신 테스트
HelloActor는 다음 특징을 갖습니다HelloActor:
"Hello"메시지를 받으면"Kotlin"응답을 보낸다.HelloCommand와HelloResponse를 주고받음
여기서 검증하는 것:
Ask Pattern을 통한 응답 받기
다양한 방식 (Coroutine, WebFlux(Mono), CompletableFuture, Java Stream, Akka Stream)으로 액터에게
ask보내고 응답받음결국 모두 "Kotlin" 이라는 응답 메시지를 받아야 성공
Pekko의
AskPattern.ask메소드를 활용하고,Duration타임아웃 설정도 적용함
testActorAskByCoroutine
@Test
fun testActorAskByCoroutine() = runBlocking {
// Test using Actor Ask pattern with coroutines
val actorRef = actorSystem.spawn(HelloActor.create())
var response = AskPattern.ask(
actorRef,
{ replyTo: ActorRef<HelloCommand> -> Hello("Hello", replyTo) },
Duration.ofSeconds(3),
actorSystem.scheduler()
).toCompletableFuture().await()
assertEquals(HelloResponse("Kotlin"), response)
}
testActorAskByWebflux
@Test
fun testActorAskByWebflux() {
val actorRef = actorSystem.spawn(HelloActor.create())
val result = Mono.just("Hello")
.flatMap { input ->
Mono.fromFuture(
AskPattern.ask(
actorRef,
{ replyTo: ActorRef<HelloCommand> -> Hello(input, replyTo) },
Duration.ofSeconds(3),
actorSystem.scheduler()
).toCompletableFuture()
)
}
.map { response -> (response as HelloResponse).message }
.block()
assertEquals("Kotlin", result)
}
testActorAskByCompletableFuture
@Test
fun testActorAskByCompletableFuture() {
val actorRef = actorSystem.spawn(HelloActor.create())
val future = CompletableFuture.supplyAsync { "Hello" }
.thenCompose { input ->
AskPattern.ask(
actorRef,
{ replyTo: ActorRef<HelloCommand> -> Hello(input, replyTo) },
Duration.ofSeconds(3),
actorSystem.scheduler()
).toCompletableFuture()
}
.thenApply { response -> (response as HelloResponse).message }
val result = future.get()
assertEquals("Kotlin", result)
}
testActorAskByJavaStream
@Test
fun testActorAskByJavaStream() {
val actorRef = actorSystem.spawn(HelloActor.create())
val result = listOf("Hello")
.stream()
.map { input ->
AskPattern.ask(
actorRef,
{ replyTo: ActorRef<HelloCommand> -> Hello(input, replyTo) },
Duration.ofSeconds(3),
actorSystem.scheduler()
).toCompletableFuture().join()
}
.map { response -> (response as HelloResponse).message }
.findFirst()
.orElse("")
assertEquals("Kotlin", result)
}
@Test
fun testActorAskByAkkaStream() {
val actorRef = actorSystem.spawn(HelloActor.create())
val materializer = Materializer.createMaterializer(actorSystem.system())
val source = Source.single("Hello")
.mapAsync(1) { input ->
AskPattern.ask(
actorRef,
{ replyTo: ActorRef<HelloCommand> -> Hello(input, replyTo) },
Duration.ofSeconds(3),
actorSystem.scheduler()
).toCompletableFuture()
}
.map { response -> (response as HelloResponse).message }
.runWith(Sink.head(), materializer)
val result = source.toCompletableFuture().get()
assertEquals("Kotlin", result)
}testActorAskByAkkaStreamWithCoroutine
@Test
fun testActorAskByAkkaStreamWithCoroutine() = runBlocking {
val actorRef = actorSystem.spawn(HelloActor.create())
val materializer = Materializer.createMaterializer(actorSystem.system())
val source = Source.single("Hello")
.mapAsync(1) { input ->
AskPattern.ask(
actorRef,
{ replyTo: ActorRef<HelloCommand> -> Hello(input, replyTo) },
Duration.ofSeconds(3),
actorSystem.scheduler()
).toCompletableFuture()
}
.map { response -> (response as HelloResponse).message }
.runWith(Sink.head(), materializer)
val result = source.await()
assertEquals("Kotlin", result)
}
(4) 코루틴 액터 (KHelloActor) 테스트
추가로 만든 **KHelloActor**는 별도 코루틴 기반 액터입니다.
특징:
내부적으로
Channel로 메시지를 수신코루틴을 통해 메시지를 받아서 응답 채널로 결과 전송
- Akka(Pekko)가 지원하는 Actor의 최소기능만 제공합니다.
JVM에서 지원하는 Thread보다 더 가벼운 경량화 실행모델
Java 21의 Virtual Thread 가 그러한 목적이며 스레드를 만들어야할 일이 생길때 대응할수 있습니다. -Java 20이하를 기본채택했을때
테스트 흐름:
KHelloActor를 생성하고start()(코루틴 Job 시작)Channel을 통해KHello메시지를 보냄응답으로
KHelloResponse를 수신결과를 검증한 후 리소스 정리 (
close및cancel)
testKHelloActor
@Test
fun testKHelloActor() = runBlocking {
// KHelloActor 생성 및 시작
val actor = KHelloActor()
val replyChannel = Channel<KHelloResponse>()
val job = actor.start() // start 메서드가 Job을 반환하도록 수정
try {
// 메시지 전송 및 응답 수신
withTimeout(3000) { // 타임아웃 설정
actor.send(KHello("Hello", replyChannel))
val response = replyChannel.receive()
// 응답 검증
assertEquals("Kotlin", response.message)
}
} finally {
// 리소스 정리
replyChannel.close()
actor.close()
job.cancel() // 액터의 코루틴 종료
}
}
🧩 3. 관련 클래스/구성요소 요약
| 클래스 | 설명 |
|---|---|
HelloActor | Pekko 기반의 기본 액터 (HelloCommand 수신, Kotlin 응답) |
BulkProcessor | (참조만) 대량 이벤트를 버퍼링하여 플러시하는 액터BulkProcessor |
HelloRouter | 라우팅 풀(RoundRobin)로 여러 액터에 분산 처리하는 액터HelloRouter |
KHelloActor | 코루틴 기반 간단한 액터 (Virtual Thread 스타일 대응 실험) |
각 클래스에 대해서 별도 HelloActorTest.kt, BulkProcessorTest.kt, HelloRouterTest.kt 테스트 파일로도 검증을 하고 있었습니다HelloActorTestBulkProcessorTestHelloRouterTest.
🧩 4. 요약
✅ 이 테스트 파일은
"비동기/동시성" 기술과 "액터모델" 메시지 송수신을 다양한 방법으로 실습하고 검증하는 포괄적 예제입니다.
✅ Kotlin Coroutine, Java CompletableFuture, Reactor Mono/Flux, Akka Stream, Pekko Actor Ask Pattern 전부를 경험할 수 있도록 구성되어 있습니다.
✅ 특히 Kotlin 기반 경량 Actor 모델 (KHelloActor)은 앞으로 Virtual Thread를 고려한 차세대 패턴을 시도하고 있다는 점이 매우 흥미로운 부분입니다.