R2DBCを導入しようとした話

1. R2DBCΛ ಋೖ͠Α͏ͱͨ͠࿩ 2023/09/29ʢۚʣ αʔόʔαΠυKotlin LTେձ vol.10 גࣜձࣾϚωʔϑΥϫʔυ Task
2. Task / Tasuku Nakagawa 2023೥7݄͔Βݱ৬ େࡕڌ఺ۈ຿ GitHub: T45K X (چTwitter): getupmax ࢿྉ: https://speakerdeck.com/t45k/r2dbcwodao-ru-suruhua
3. ϚωʔϑΥϫʔυΫϥ΢υ ࿈݁ձܭ
4. KotlinʢJVMʣͷ஌ݟ͕ͳ͍ঢ়ଶͰ։ൃ։࢝ →͍͔ͭ͘ͷ΍Γ͍ͨ͜ͱΛఘΊͯϦϦʔε Ұ݅ͷܖ໿ͰϢʔβ͕Ͳͬͱ૿͑Δ →ૣ͍͏ͪʹεέʔϧ͢ΔΑ͏ʹ͍ͨ͠ ϚωʔϑΥϫʔυΫϥ΢υ ࿈݁ձܭ
5. DBΞΫηεΛϊϯϒϩοΩϯάͰ΍Δ΍ͭ ↕︎ JDBC R2DBC
6. • ϋϚΓϙΠϯτ • ʢݱঢ়ͷʣղܾࡦ ӡ༻ͷ࿩͸͠·ͤΜʢ·ͩͰ͖ͳ͍ʣ R2DBCಋೖ࿥
7. • Kotlin 1.9.0 • Spring Boot 3.0.5 • Spring WebFlux • jOOQ 3.7.5 • MySQL ελοΫ
8. ϦϙδτϦ
9. @Repository class Repository(private val ctx: DSLContext) { fun findById(id: Long): Record? = ctx.selectFrom(TABLE) .where(ID.eq(id)) .fetchOne() } DSLͰDB໰͍߹Θ͕ͤͰ͖ΔORM jOOQͱ͸
10. jOOQ͸R2DBCʹରԠࡁΈ https://blog.jooq.org/reactive-sql-with-jooq-3-15-and-r2dbc/
11. dependencies { runtimeOnly("io.asyncer:r2dbc-mysql:1.0.2") runtimeOnly("io.r2dbc:r2dbc-pool:1.0.1.RELEASE") // ͪ͜Β͸ݱࡏ։ൃ͕ࢭ·͍ͬͯΔ // runtimeOnly("dev.miku:r2dbc-mysql:0.8.2.RELEASE") } ґଘ build.gradle.kts
12. spring: r2dbc: url: r2dbc://{jdbcͱಉ͡ઃఆ} username: username password: password pool: enabled: true ઃఆ application.yaml
13. @Configuration class JooqConfiguration( private val connectionFactory: ConnectionFactory) { @Bean fun dslContext(): DSLContext = DSL.using(connectionFactory).dsl() } jOOQ Config
14. jOOQͰR2DBCΛ ࢖͏४උ͸੔ͬͨ
15. jOOQ 3.17͔Β coroutineΛαϙʔτ https://www.jooq.org/doc/3.17/manual/sql-building/kotlin-sql-building/kotlin-coroutines/
16. @Repository class Repository(private val ctx: DSLContext) { suspend fun findById(id: Long): Record? = ctx.selectFrom(TABLE) .where(ID.eq(id)) .awaitFirstOrNull() } jOOQ 3.17ͷมߋ఺ ΫΤϦ͕ PublisherΛฦ͢ Coroutineʹ ม׵Մೳ
17. ͳ͍
18. jOOQ 3.17.5࣌఺Ͱ͸ ࣮૷͞Ε͍ͯͳ͍ ʢηϚϯςΟοΫόʔδϣχϯά...ʣ
19. jOOQ 3.17.15 or 3.18ܥΛ ࢖͏ͱղܾ
20. fetch/fetchOneΛ ଟ༻͍ͯ͠Δ
21. fetch/fetchOne͸ ϒϩοΩϯά͢Δ શͯΛselectʹ ॻ͖׵͑Δͷ΋໘౗
22. fetch/fetchOneޓ׵ͷ ϝιουΛ४උͯ͠Ұׅஔ׵ suspend inline fun <reified R : TableRecord<*>, T : TableImpl<R>> DSLContext.nonBlockingFetchOne( tableImpl: T, vararg condition: Condition ): R? = this.select() .from(tableImpl) .where(*condition) .awaitFirstOrNull() ?.map { it.into(R::class.java) }
23. τϥϯβΫγϣϯ
24. ݱঢ়ɺR2DBCͰ @Transaction͸ ࢖͑ͳͦ͞͏
25. R2DBC༻ͷ τϥϯβΫγϣϯ؅ཧ
26. class UseCase( private val tx: TransactionalOperator, private val repo: Repository, ) { suspend fun call() = tx.execute(mono { repo.save() }) } Spring Frameworkۘ੡ Ҿ਺͕Mono TransactionOperator
27. class UseCase( private val ctx: DSLContext, private val repo: Repository, ) { suspend fun call() = ctx.transactionCoroutine { repo.save() } } jOOQۘ੡ DSLContext͔Βੜ͍͑ͯΔ transactionCoroutine
28. લऀ͸jOOQඇରԠ ࣮࣭ޙऀҰ୒
29. class UseCase( private val ctx: DSLContext, private val repo: Repository, ) { suspend fun call() = ctx.transactionCoroutine { repo.save() } } jOOQۘ੡ DSLContext͔Βੜ͍͑ͯΔ transactionCoroutine
30. class UseCase( private val ctx: DSLContext, private val repo: Repository, ) { suspend fun call() = ctx.transactionCoroutine { repo.save() } } jOOQۘ੡ DSLContext͔Βੜ͍͑ͯΔ transactionCoroutine ͜Ε͚ͩͰ͸ޮ͔ͳ͍
31. class UseCase( private val ctx: DSLContext, private val repo: Repository, ) { suspend fun call() = ctx.transactionCoroutine { config -> repo.save(config.dsl()) } } ίʔϧόοΫ಺ͰDSLContextΛऔಘ ͦΕʹରͯ͠ΫΤϦΛ࣮ߦ͠ͳ͍ͱ͍͚ͳ͍ transactionCoroutineͷ᠘
32. όέπϦϨʔʁ class UseCase( private val ctx: DSLContext, private val service: Service, ) { suspend fun call() = ctx.transactionCoroutine { config -> service.do(config.dsl()) } } class Service( private val repo: Repository, ) { suspend fun do(ctx: DSLContext) { repo.save(ctx) } }
33. όέπϦϨʔʁ class UseCase( private val ctx: DSLContext, private val service: Service, ) { suspend fun call() = ctx.transactionCoroutine { config -> service.do(config.dsl()) } } class Service( private val repo: Repository, ) { suspend fun do(ctx: DSLContext) { repo.save(ctx) } } όέπϦϨʔ όέπϦϨʔ
34. ίʔϧόοΫ಺Ͱݺ͹Εͨ Repository͸ DSLContextΛ ੾Γସ͍͑ͨ
35. CoroutineContextͰ ࣋ͪճ͢
36. class TransactionElement(val ctx: DSLContext) : AbstractCoroutineContextElement(TransactionElement) { companion object Key : CoroutineContext.Key<TransactionElement> } DSLContextΛอ࣋͢Δ CoroutineContext
37. @Component class TransactionManager(private val ctx: DSLContext) { suspend fun <T> execute(block: suspend () -> T): T = ctx.transactionCoroutine { config -> withContext(coroutineContext + TransactionElement(config.dsl())) { block() } } } } ϚωʔδϟͰContextΛ߹੒
38. @Repository class Repository(private val ctx: DSLContext) { suspend fun getCtx() = coroutineContext[TransactionElement.Key]?.ctx ?: this.ctx suspend fun save() { getCtx()... // CRUDॲཧ } } RepoͰ࢖͏ctxΛ੾Γସ͑
39. AOP
40. @Component @Aspect class CommonProcess( private val repo: Repository, ) { @Before("execution(* *Query.*(..))") fun check(joinPoint: JoinPoint): Any? { repo.find() ?: throw RuntimeException() } } AOPͰڞ௨ॲཧΛ͍ͯ͠Δ
41. @Component @Aspect class CommonProcess( private val repo: Repository, ) { @Before("execution(* *Query.*(..))") fun check(joinPoint: JoinPoint): Any? { repo.find() ?: throw RuntimeException() } } AOPͰڞ௨ॲཧΛ͍ͯ͠Δ ϦϙδτϦͷϝιου͕ suspendؔ਺ʹͳͬͨ
42. @Component @Aspect class CommonProcess( private val repo: Repository, ) { @Before("execution(* *Query.*(..))") fun check(joinPoint: JoinPoint): Any? { runBlocking { repo.find() } ?: throw RuntimeException() } } runBlocking͢Δʁ
43. @Component @Aspect class CommonProcess( private val repo: Repository, ) { @Before("execution(* *Query.*(..))") fun check(joinPoint: JoinPoint): Any? { runBlocking { repo.find() } ?: throw RuntimeException() } } runBlocking͢Δʁ ελοΫ͢Δ ʢݪҼෆ໌ʣ
44. Coroutineͷ࣮ߦʹ ׂΓࠐ·͍ͤͨ
45. suspend fun hoge(value: String){ } public static final Object hoge( String value, Continuation $completion) { return Unit.INSTANCE; } suspendؔ਺͸ Java͔Β͜͏ݟ͑Δ
46. suspend fun hoge(value: String){ } public static final Object hoge( String value, Continuation $completion) { return Unit.INSTANCE; } suspendؔ਺͸ Java͔Β͜͏ݟ͑Δ Continuation͕ Coroutineͷ͋Ε͜ΕΛ࣋ͭ
47. @Component @Aspect class CommonProcess( private val repo: Repository, ) { @Around("execution(* *Query.*(..))") fun check(pjp: ProceedingJoinPoint): Any? { val con = pjp.args.last() as Continuation<Any?> return suspend { repo.find() ?: throw RuntimeException() suspendCoroutineUninterceptedOrReturn<Any?> { con2 -> val args = pjp.args.sliceArray(0..<pjp.args.size - 1) joinPoint.proceed(args + con2) } }.startCoroutineUninterceptedOrReturn(con) } }
48. @Component @Aspect class CommonProcess( private val repo: Repository, ) { @Around("execution(* *Query.*(..))") fun check(pjp: ProceedingJoinPoint): Any? { val con = pjp.args.last() as Continuation<Any?> return suspend { repo.find() ?: throw RuntimeException() suspendCoroutineUninterceptedOrReturn<Any?> { con2 -> val args = pjp.args.sliceArray(0..<pjp.args.size - 1) joinPoint.proceed(args + con2) } }.startCoroutineUninterceptedOrReturn(con) } } JoinPoint͔Β ContinuationΛύΫΔ
49. @Component @Aspect class CommonProcess( private val repo: Repository, ) { @Around("execution(* *Query.*(..))") fun check(pjp: ProceedingJoinPoint): Any? { val con = pjp.args.last() as Continuation<Any?> return suspend { repo.find() ?: throw RuntimeException() suspendCoroutineUninterceptedOrReturn<Any?> { con2 -> val args = pjp.args.sliceArray(0..<pjp.args.size - 1) joinPoint.proceed(args + con2) } }.startCoroutineUninterceptedOrReturn(con) } } ύΫͬͨContinuation্Ͱ DBίʔϧ
50. @Component @Aspect class CommonProcess( private val repo: Repository, ) { @Around("execution(* *Query.*(..))") fun check(pjp: ProceedingJoinPoint): Any? { val con = pjp.args.last() as Continuation<Any?> return suspend { repo.find() ?: throw RuntimeException() suspendCoroutineUninterceptedOrReturn<Any?> { con2 -> val args = pjp.args.sliceArray(0..<pjp.args.size - 1) joinPoint.proceed(args + con2) } }.startCoroutineUninterceptedOrReturn(con) } } ͦͷ··JoinPoint࣮ߦ
51. Spring Framework 6.1͔Β αϙʔτ͞ΕΔ໛༷😇 https://github.com/spring-projects/spring-framework/issues/22462
52. ͦͷଞ
53. • ஈ֊తʹमਖ਼͍͖͍ͯͨ͠ • ݁߹ςετͰɺJDBCͰDBΛݕূ͍ͨ͠ • suspendؔ਺ʹͰ͖ͳ͍ؔ਺͕͋Δ JDBCͱR2DBCͷซ༻
54. @Configuration class JooqConfiguration( private val connectionFactory: ConnectionFactory) { @Bean fun dslContext(): DSLContext = DSL.using(connectionFactory).dsl() @Bean("jdbcDSLContext") fun jdbcDslContext( @Value("\${spring.datasource.url}") url: String, @Value("\${spring.datasource.username}") username: String, @Value("\${spring.datasource.password}") password: String, ): DSLContext = DefaultDSLContext( DataSourceBuilder.create().url(url).username(username) .password(password).build(), SQLDialect.MYSQL) } ࣗલͰ༻ҙ͢Δ
55. @Repository class Repository( private val ctx: DSLContext, @Qualifier("jdbcDSLContext") private val jdbcCtx: DSLContext, ) { fun findById(id: Long): Record? = ctx… fun blockingFindById(id: Long): Record? = jdbcCtx… } ϦϙδτϦ
56. ىಈ࣌ʹUnknownHostException NettyͷDNSϦκϧό໰୊ Caused by: java.net.UnknownHostException: Failed to resolve 'my_db:3306' [A(1)] after 2 queries at io.netty.resolver.dns.DnsResolveContext.finishResolve (DnsResolveContext.java:1088)
57. DBͷϗετ໊ʹ”_”ؚ͕·Εͯ͸͍͚ͳ͍ Docker Compose౳࢖͍ͬͯΔ৔߹஫ҙ NettyͷDNSϦκϧό໰୊
58. • jOOQ͸৽͍͠όʔδϣϯΛ࢖͏ • CoroutineContextΛ׆༻ͯ͠
 τϥϯβΫγϣϯΛ؅ཧ͢Δ • AOPͰ͸ContinuationΛ࢖͍ճ͢ • JDBCͱซ༻ͯ͠ஈ֊తʹमਖ਼͢Δ ·ͱΊ
59. • QA • ϦϦʔεޙελοΫ͠ͳ͍͜ͱΛفΔ • ύϑΥʔϚϯεܭଌ͍ͨ͠ ࠓޙ
60. WE’RE HIRING!! https://hrmos.co/pages/moneyforward/jobs?category=1666322478647443458,1666323214451404802