Share
Embed
Transcript
47 slides
-
Agenda • Virtual Threads ѐਃ • Rules & Best practices
• Spring Boot with Virtual Threads • Legacy Code ߸҃ Generated by ChatGPT -
Virtual Threads (Loom project) ۆ • Virtual Threads ౠ •
Lightweight Threads ( Not Fiber, Not Coroutines ) • Virtual Thread Per Request • Non-Blocking ਗ • ӝઓ Platform Thread৬ ഐജؼ ࣻ ب۾ ೠ • Legacy Code ߸ചহ ࢎਊ оמ೧ঠ ೠ • Coroutines, Reactive Programming ۢ ѐߊ ߑध ܳ ജ೧ঠ ೞח Ѫ ইפ • Virtual threads are syntatic sugar for reactive programming • ࠺زӝ ٘ܳ sequential code ۽ അ оמ — Readable Code -
Quiz Platform Threads vs Virtual Threads - Blocking ? @Test
fun `Platform Thread ۽ Blocking ٘ प೯ೞӝ`() { val threads = List(THREAD_SIZE / 100) { Thread.ofPlatform().start { Thread.sleep(1000) log.debug { "Platform Thread $it" } } } threads.forEach { it.join() } } @Test fun `Virtual Thread ۽ Blocking ٘ प೯ೞӝ`() { val threads = List(THREAD_SIZE) { Thread.ofVirtual().start { Thread.sleep(1000) log.debug { "Virtual Thread $it" } } } threads.forEach { it.join() } } -
Quiz Platform Threads vs Virtual Threads - Blocking ? @Test
fun `Platform Thread ۽ Blocking ٘ प೯ೞӝ`() { val threads = List(THREAD_SIZE / 100) { Thread.ofPlatform().start { Thread.sleep(1000) log.debug { "Platform Thread $it" } } } threads.forEach { it.join() } } @Test fun `Virtual Thread ۽ Blocking ٘ प೯ೞӝ`() { val threads = List(THREAD_SIZE) { Thread.ofVirtual().start { Thread.sleep(1000) log.debug { "Virtual Thread $it" } } } threads.forEach { it.join() } } -
Quiz Platform Threads vs Virtual Threads - Blocking ? @Test
fun `Platform Thread ۽ Blocking ٘ प೯ೞӝ`() { val threads = List(THREAD_SIZE / 100) { Thread.ofPlatform().start { Thread.sleep(1000) log.debug { "Platform Thread $it" } } } threads.forEach { it.join() } } @Test fun `Virtual Thread ۽ Blocking ٘ प೯ೞӝ`() { val threads = List(THREAD_SIZE) { Thread.ofVirtual().start { Thread.sleep(1000) log.debug { "Virtual Thread $it" } } } threads.forEach { it.join() } } 100,000 * 1 sec / core + logging -
Quiz Platform Threads vs Virtual Threads - Blocking ? @Test
fun `Platform Thread ۽ Blocking ٘ प೯ೞӝ`() { val threads = List(THREAD_SIZE / 100) { Thread.ofPlatform().start { Thread.sleep(1000) log.debug { "Platform Thread $it" } } } threads.forEach { it.join() } } @Test fun `Virtual Thread ۽ Blocking ٘ प೯ೞӝ`() { val threads = List(THREAD_SIZE) { Thread.ofVirtual().start { Thread.sleep(1000) log.debug { "Virtual Thread $it" } } } threads.forEach { it.join() } } 100,000 * 1 sec / core + logging -
Quiz Platform Threads vs Virtual Threads - Blocking ? @Test
fun `Platform Thread ۽ Blocking ٘ प೯ೞӝ`() { val threads = List(THREAD_SIZE / 100) { Thread.ofPlatform().start { Thread.sleep(1000) log.debug { "Platform Thread $it" } } } threads.forEach { it.join() } } @Test fun `Virtual Thread ۽ Blocking ٘ प೯ೞӝ`() { val threads = List(THREAD_SIZE) { Thread.ofVirtual().start { Thread.sleep(1000) log.debug { "Virtual Thread $it" } } } threads.forEach { it.join() } } 100,000 * 1 sec / core + logging Under 3 seconds -
JVM Platform Threads ز ߑध Application Thread Pool Platform Thread
Platform Thread Platform Thread Platform Thread OS OS Thread OS Thread OS Thread OS Thread -
JVM Virtual Threads ز ߑध Application Carrier Thread Carrier Thread
Platform Thread Carrier Thread OS OS Thread OS Thread OS Thread OS Thread Virtual Thread Scheduler Virtual Thread Virtual Thread Virtual Thread Virtual Thread Virtual Thread -
JVM Virtual Threads ز ߑध Application Carrier Thread Carrier Thread
Platform Thread Carrier Thread OS OS Thread OS Thread OS Thread OS Thread Virtual Thread Scheduler Virtual Thread Virtual Thread -
JVM Virtual Threads ز ߑध Application Carrier Thread Carrier Thread
Platform Thread Carrier Thread OS OS Thread OS Thread OS Thread OS Thread Virtual Thread Scheduler Virtual Thread Virtual Thread -
JVM Virtual Threads ز ߑध Application Carrier Thread Carrier Thread
Platform Thread Carrier Thread OS OS Thread OS Thread OS Thread OS Thread Virtual Thread Scheduler Virtual Thread Virtual Thread -
JVM Virtual Threads ز ߑध Application Carrier Thread Carrier Thread
Platform Thread Carrier Thread OS OS Thread OS Thread OS Thread OS Thread Virtual Thread Scheduler Virtual Thread Virtual Thread -
JVM Virtual Threads ز ߑध Application Carrier Thread Carrier Thread
Platform Thread Carrier Thread OS OS Thread OS Thread OS Thread OS Thread Virtual Thread Scheduler Virtual Thread Virtual Thread Blocking -
JVM Virtual Threads ز ߑध Application Carrier Thread Carrier Thread
Platform Thread Carrier Thread OS OS Thread OS Thread OS Thread OS Thread Virtual Thread Scheduler Virtual Thread Virtual Thread Blocking -
JVM Virtual Threads ز ߑध Application Carrier Thread Carrier Thread
Platform Thread Carrier Thread OS OS Thread OS Thread OS Thread OS Thread Virtual Thread Scheduler Virtual Thread Virtual Thread Blocking -
Platform Thread Virtual Thread Coroutines Startup time > 1,000 us
1~10 us 1~10 us Metadata Size ~ 2kb 200~300 byte Memory ܻ ೡػ Stack ࢎਊ (1MB) ਃ द Heap ࢎਊ (~1,400 bytes) ਃ द Heap ࢎਊ (~300 bytes) Context switching cost 1 ~ 10 us (ழօ) ~ 2 us ~ 1 us Numbers < 5,000 Millions Millions Best cases CPU Bounded Per Request Legacy Code High-concurrent code Event based Systems Structured concurrency and cancellation -
Rule 1 - Create Threads val builder = Thread.ofPlatform() .daemon(false)
.priority(10) .stackSize(1024) .name("platform-thread") .inheritInheritableThreadLocals(false) .uncaughtExceptionHandler { thread, ex -> print("Thread[$thread] failed with exception: $ex") } val thread = builder.unstarted { print("Platform Thread") } val builder = Thread.ofVirtual() .name("virtual-thread") .inheritInheritableThreadLocals(false) .uncaughtExceptionHandler { thread, ex -> log.error(ex) { "Thread[$thread] failed with exception." } } val thread = builder.unstarted { println("Unstarted Virtual Thread") } -
Rule 2 - Do Not use CompletableFuture For Legacy synchronous
code fun `زӝ ٘ܳ CompletableFuture ۽ प೯ೞӝ`() { val startMs = System.currentTimeMillis() CompletableFuture .supplyAsync { readPriceInEur() } .thenCombine( CompletableFuture.supplyAsync { readExchangeRateEurToUsd() }) { price, rate -> price * rate } .thenCompose { amount -> CompletableFuture.supplyAsync { amount * (1.0F + readTax(amount)) } } .whenComplete { grossAmountInUsd, error -> if (error == null) { grossAmountInUsd.toInt() shouldBeEqualTo 108 } else { fail(error) } } .get() // য૰ٚ ৈӝࢲ Blocking ػ (Non-Blocking ইפ) val durationMs = System.currentTimeMillis() - startMs durationMs shouldBeInRange 800L..900L } -
Rule 2 - Do use Virtual Threads For Legacy synchronous
code fun `زӝ ٘ܳ Virtual Thread ۽ प೯ೞӝ`() { Executors.newVirtualThreadPerTaskExecutor().use { executor -> val startMs = System.currentTimeMillis() val priceInEur = executor.submit<Int> { readPriceInEur() } val exchangeRateEuroToUsd = executor.submit<Float> { readExchangeRateEurToUsd() } val netAmountInUsd = priceInEur.get() * exchangeRateEuroToUsd.get() val tax = executor.submit<Float> { readTax(netAmountInUsd) } val grossAmountInUsd = netAmountInUsd * (1.0F + tax.get()) grossAmountInUsd.toInt() shouldBeEqualTo 108 val durationMs = System.currentTimeMillis() - startMs durationMs shouldBeInRange 800L..900L } } -
Cf. Use Coroutines Change Legacy code to suspended fun `Suspend
ೣࣻܳ Coroutines ജ҃ীࢲ प೯ೞӝ`() = runSuspendTest(Dispatchers.Default) { val startMs = System.currentTimeMillis() val priceInEur = async { readPriceInEurAwait() } val exchangeRateEuroToUsd = async { readExchangeRateEurToUsdAwait() } val netAmountInUsd = priceInEur.await() * exchangeRateEuroToUsd.await() val tax = async { readTax(netAmountInUsd) } val grossAmountInUsd = netAmountInUsd * (1.0F + tax.await()) grossAmountInUsd.toInt() shouldBeEqualTo 108 val durationMs = System.currentTimeMillis() - startMs durationMs shouldBeInRange 800L..900L } -
Rule 3 - Do not use Thread Pool Virtual Threads
ࢎਊ दীח Thread Pool ਸ ࢎਊೞ ݃ۄ. Thread ࣻܳ ઁೠೡ ਬо হ fun `࠺୶ୌ - ThreadPool ীࢲ Virtual Thread ࢤࢿೞӝ`() { Executors.newCachedThreadPool(Thread.ofVirtual().factory()).use { executor -> executor.submit { Thread.sleep(1000) log.debug { "1 run ${Thread.currentThread()}" } } executor.submit { Thread.sleep(1000) log.debug { "2 run ${Thread.currentThread()}" } } } } -
Rule 3 - Do use newThreadPerTaskExecutor Virtual Threads ח Million
ױਤ۽ ࢤࢿೞח Ѫ @Test fun `୶ୌ - ThreadPerTaskExecutor ۽ Virtual Thread ࢤࢿೞӝ`() { Executors.newThreadPerTaskExecutor(Thread.ofVirtual().factory()).use { executor -> executor.submit { Thread.sleep(1000) log.debug { "1 run ${Thread.currentThread()}" } } executor.submit { Thread.sleep(1000) log.debug { "2 run ${Thread.currentThread()}" } } } } -
Rule 4 - Do not use FixedThreadPool for concurrency fun
`࠺୶ୌ - FixedThreadPoolਸ ࢎਊೞৈ زदࢿ ઁযೞӝ`() { val results = ConcurrentLinkedQueue<Future<String>>() Executors.newFixedThreadPool(8, Thread.ofVirtual().factory()).use { executor -> val futures = List(100) { index -> executor.submit { log.debug { "Start run task[$index]" } val result = executor.submit<String> { sharedResource() } log.debug { "Finish run task[$index]" } results.add(result) } } futures.forEach { it.get() } results.size shouldBeEqualTo 100 } } -
Rule 4 - Do use Semaphore for Concurrency Local /
Distributed Semaphore fun `୶ୌ - Semaphoreܳ ਊೞৈ زदࢿ ઁযೞӝ`() { val results = ConcurrentLinkedQueue<String>() Executors.newVirtualThreadPerTaskExecutor().use { executor -> val futures = List(100) { index -> executor.submit { log.debug { "Start run task[$index]" } val result = useSemaphoreToLimitConcurrency() log.debug { "Finish run task[$index]" } results.add(result) } } futures.forEach { it.get() } results.size shouldBeEqualTo 100 } } private val semaphore = Semaphore(8) private fun useSemaphoreToLimitConcurrency(): String { semaphore.acquire() return try { sharedResource() } finally { semaphore.release() } } -
Rule 5 - Do not use ThreadLocal private val threadLocal
= InheritableThreadLocal<String>() @Test fun `࠺୶ୌ - ThreadLocal ߸ࣻܳ ࢎਊೞӝ`() { threadLocal.set("zero") threadLocal.get() shouldBeEqualTo “zero" threadLocal.set("one") threadLocal.get() shouldBeEqualTo "one" val childThread = Thread { threadLocal.get() shouldBeEqualTo "one" } childThread.start() childThread.join() threadLocal.remove() threadLocal.get().shouldBeNull() } -
Rule 5 - Do use ScopedValue ThreadLocal ਸ ࢎਊೞ ݈Ҋ,
ScopedValue ܳ ࢎਊೞࣁਃ private val scopedValue = ScopedValue.newInstance<String>() @Test fun `୶ୌ - ScopedValue ࢎਊೞӝ`() { ScopedValue.runWhere(scopedValue, "zero") { scopedValue.get() shouldBeEqualTo "zero" ScopedValue.runWhere(scopedValue, "one") { scopedValue.get() shouldBeEqualTo "one" } scopedValue.get() shouldBeEqualTo "zero" StructuredTaskScope.ShutdownOnFailure().use { scope -> // Scope উীࢲ sub taskܳ ࢤࢿפ. scope.fork { scopedValue.get() shouldBeEqualTo "zero" -1 } scope.join().throwIfFailed() } } } -
Rule 6 - Do use ReentranceLock instead of synchronized Synchronized
ח Carrier Thread ܳ Blocking פ private val lock = ReentrantLock() @Test fun `୶ୌ - ܻࣗझܳ ةਵ۽ ࢎਊೡ ٸ ReentrantLockਸ ਊೞࣁਃ - Kotlin withLock ೣࣻ`() { virtualFuture { lock.withLock { exclusiveResource() } }.await() } private fun exclusiveResource(): String { Thread.sleep(100) return "result" } How to solve the pinning problem in Java virtual threads -
Rule 7 - Do Structured Concurrency StructuredTaskScope.ShutdownOnFailure /** * ࠺زӝ
٘ܳ ߽۳۽ प೯ೞҊ, ݽٚ স ࢿҕਵ۽ ৮ܐغݶ Ѿҗܳ ߈ജפ. */ fun prepareDish(): Dish = structuredTaskScopeAll { scope -> println("prepare Dish ...") val pasta = scope.fork { Thread.sleep(100) preparePasta() } val sauce = scope.fork { Thread.sleep(200) makeSaurce() } Thread.sleep(5) scope.join().throwIfFailed() Dish(pasta.get(), sauce.get()) } -
Rule 7 - Do Structured Concurrency StructuredTaskScope.ShutdownOnFailure /** * ࠺زӝ
٘ܳ ߽۳۽ प೯ೞҊ, ݽٚ স ࢿҕਵ۽ ৮ܐغݶ Ѿҗܳ ߈ജפ. */ fun prepareDish(): Dish = structuredTaskScopeAll { scope -> println("prepare Dish ...") val pasta = scope.fork { Thread.sleep(100) preparePasta() } val sauce = scope.fork { Thread.sleep(200) makeSaurce() } Thread.sleep(5) scope.join().throwIfFailed() Dish(pasta.get(), sauce.get()) } inline fun <T> structuredTaskScopeAll( name: String? = null, factory: ThreadFactory = Thread.ofVirtual().factory(), block: (scope: StructuredTaskScope.ShutdownOnFailure) -> T, ): T { return StructuredTaskScope.ShutdownOnFailure(name, factory).use { scope -> block(scope) } } -
Rule 7 - Do Structured Concurrency StructuredTaskScope.ShutdownOnSuccess fun `structured task
scope on success`() { // Subtask ٜ ೞաۄب ࢿҕೞݶ, աݠ Subtask ٜ ஂࣗೞҊ, Ѿҗܳ ߈ജפ. // ݅ড ࢿҕೠ Ѫ হݶ ExecutionException ਸ ߈ജפ. val pasta = structuredTaskScopeFirst<Pasta> { scope -> val subtask1 = scope.fork { Thread.sleep(100) preparePasta() } val subtask2 = scope.fork { Thread.sleep(200) preparePasta() } scope.join() subtask1.state() shouldBeEqualTo StructuredTaskScope.Subtask.State.SUCCESS subtask2.state() shouldBeEqualTo StructuredTaskScope.Subtask.State.UNAVAILABLE scope.result { ExecutionException(it) } } } -
Rule 7 - Do Structured Concurrency StructuredTaskScope.ShutdownOnSuccess fun `structured task
scope on success`() { // Subtask ٜ ೞաۄب ࢿҕೞݶ, աݠ Subtask ٜ ஂࣗೞҊ, Ѿҗܳ ߈ജפ. // ݅ড ࢿҕೠ Ѫ হݶ ExecutionException ਸ ߈ജפ. val pasta = structuredTaskScopeFirst<Pasta> { scope -> val subtask1 = scope.fork { Thread.sleep(100) preparePasta() } val subtask2 = scope.fork { Thread.sleep(200) preparePasta() } scope.join() subtask1.state() shouldBeEqualTo StructuredTaskScope.Subtask.State.SUCCESS subtask2.state() shouldBeEqualTo StructuredTaskScope.Subtask.State.UNAVAILABLE scope.result { ExecutionException(it) } } } inline fun <T> structuredTaskScopeFirst( name: String? = null, factory: ThreadFactory = Thread.ofVirtual().factory(), block: (scope: StructuredTaskScope.ShutdownOnSuccess<T>) -> T, ): T { return StructuredTaskScope.ShutdownOnSuccess<T>(name, factory).use { scope -> block(scope) } } -
Rule 8 - Test Virtual Threads @Test fun `get all
actors in multiple virtual threads`() { VirtualthreadTester() .numThreads(Runtimex.availableProcessors * 2) .roundsPerThread(4) .add { transaction(db) { val actors = Actor.all().toList() actors.shouldNotBeEmpty() } } .run() } Add ೣࣻী য ٘ ࠶۟ਸ CPU Core ࣻ * 2 * 4 ഥ ݅ఀ زदী ࣻ೯ೠ -
Spring Boot MVC with Virtual Threads @Configuration class TomcatConfig {
/** * Tomcat ProtocolHandler executor ܳ Virtual Thread ܳ ࢎਊೞח Executorܳ ࢎਊೞب۾ ࢸ */ @Bean fun protocolHandlerVirtualThreadExecutorCustomizer(): TomcatProtocolHandlerCustomizer<*> { return TomcatProtocolHandlerCustomizer<ProtocolHandler> { protocolHandler -> protocolHandler.executor = Executors.newVirtualThreadPerTaskExecutor() } } } server: tomcat: threads: max: 8000. # ӝࠄ: 200, Virtual Threads ח ݅ױਤ ࢚ਵ۽ ೧ب ޖߑೞ min-spare: 20 application.yml -
Spring Boot MVC Async Tasks /** * `@Async` য֢ప࣌ ਊػ
ݫࣗ٘ܳ Virtual Threaedܳ ਊೞৈ ࠺زӝ۽ प೯ೞӝ ਤೠ ࢸ */ @Configuration @EnableAsync class AsyncConfig { @Bean(TaskExecutionAutoConfiguration.APPLICATION_TASK_EXECUTOR_BEAN_NAME) fun asyncTaskExecutor(): AsyncTaskExecutor { val factory = Thread.ofVirtual().name("async-vt-exec-", 0).factory() return TaskExecutorAdapter(Executors.newThreadPerTaskExecutor(factory)) } } -
RestController use Virtual Threads @GetMapping("/multi") fun multipleTasks(): String { val
taskSize = 1000 // ShutdownOnFailure - ೞաۄب पಁೞݶ द ઙܐ (೯ ੋ ܲ Task ٜ ױػ) StructuredTaskScope.ShutdownOnFailure().use { scope -> repeat(taskSize) { scope.fork { Thread.sleep(1000) log.debug { "Task $it is done. (${Thread.currentThread()})" } } } scope.join().throwIfFailed() } return "Run multiple[$taskSize] tasks. (${Thread.currentThread()})" } -
ӝઓ दझమী Virtual Threads ਊೞӝ • ࢚ • IO Bounded
Operations • API ഐ (ӝઓ Feign, WebClient ഐ ߑध ߸҃) • Database IO স (୭न JDBC Driver ࢎਊ ਃ) • Ӓ ৻ IO স زӝ ߑध ژח CompletableFutureܳ ࢎਊೞח Ҕ • ઁ৻ ࢚ • CPU Bounded Operations (ஸ۩࣌ স ١) • WAS Thread Pool ߸҃ • Spring Boot embedded web server (Tomcat, Undertow) Thread Pool ਸ Virtual Threads ۽ -
Resources • Virtual Thread ӝࠄ ѐ֛ ೧ೞӝ - Naver
D2 • [Project Loom] Virtual Thread ী ࠆ(Spring) ৳חо - KakaoPay • Concurrent programming in Java with virtual threads • Java 21 Virtual Threads - Dude, Where’s My Lock? - Net fl ix blog • Embracing Virtual Threads - Spring blog • Code examples • Bluetape4k-workshop/virtualthreads