Approche impérative des données réactives sur l'exemple de Jetbrains KTor et R2DBC

Un article sur l'utilisation de l'accès réactif aux bases de données à partir de la corutine. Spring simplifie tout, mais il n'affecte pas la compréhension des processus réels de l'application. Pour démonstration, le framework KTor a été choisi (simplement parce que j'aime regarder ce que fait JetBrains), qui utilise intensivement les coroutines - de sorte que la tâche de combiner Reactive Streams et ces mêmes coroutines ajoute de l'intérêt. Dans le processus, il est devenu clair que ce qui se passait était un exemple clair de la transformation d'un jet stream incompréhensible en une programmation impérative compréhensible, sur laquelle nous avons mangé un chien. J'adore les chaînes à réaction, mais pourquoi ne pas plaire à ceux qui aiment l'ordre de l'armée?

Les applications Jet ont conquis le cœur et les nerfs de nombreux développeurs, et ces ensembles se croisent nettement. Ils planteraient encore plus sans les efforts des communautés pour adapter le pur courant d'esprit des créateurs de spécifications aux bibliothèques digestes. Cela s'est produit avec la spécification R2DBC et le framework Spring (Boot) - le développeur peut déjà voir l'API Spring Data familière avec les types de données réactifs familiers. Cependant, il y a des raisons de ne pas utiliser Spring: vous ne voulez pas de Spring et vous voulez quelque chose de nouveau. Eh bien, il reste du code hérité, mais dans ce cas, il est peu probable que vous ayez un accès réactif aux données.

Dans ce cas, vous devez regarder R2DBC sans vernis. Et il devrait être très différent de ce qui nous est proposé dans le cadre fini - tout comme JDBC est différent de Spring Data JPA. Plus de réactivité. Et la réactivité selon la spécification Reactive Streams. Et nous entendons des coroutines. Quel genre de comme l'avenir et encore les réécrire.

Vous pouvez également démarrer les coroutines manuellement à partir de la méthode principale , mais essayons d'imaginer que nous avons vraiment besoin de parallélisme et de concurrence - c'est-à-dire une charge élevée (disons, une demande par heure!) Et nous avons sérieusement décidé d'écrire ceci sans Spring. Mais ici, il s'avère, nous avons un analogue léger entièrement écrit en Kotlin, et même des créateurs de la langue elle-même, et même sur les coroutines dont nous rêvons.

Nous préparons un projet

, - ( IntelliJ IDEA).

https://start.ktor.io :

• Call Logging: ,
• Routing: (endpoint URI)
• Gson: .

, — OAuth, JWT, LDAP , .

. , IntelliJ IDEA Community Edition.

H2, build.gradle

    implementation "io.r2dbc:r2dbc-h2:0.8.1.RELEASE"

    implementation "io.r2dbc:r2dbc-pool:0.8.1.RELEASE"

Reactive Streams, R2DBC , Reactor Kotlin, :

    implementation "io.projectreactor.kotlin:reactor-kotlin-extensions:1.0.2.RELEASE"
    implementation "org.jetbrains.kotlinx:kotlinx-coroutines-reactor:1.3.5"

: JVM 1.8. , - native Java. .

build.gradle

tasks.withType(org.jetbrains.kotlin.gradle.tasks.KotlinCompile).configureEach {
    kotlinOptions {
        jvmTarget = "1.8"
    }
}

— Application.kt, . , , :

fun Application.module(testing: Boolean = false) {
    //       
initDatabase@

    //   

applicationStarted@
    //       

boringStandardInitialization@
    //   ,   
    install(ContentNegotiation) {
        gson {
        }
    }

    install(CallLogging) {
        level = Level.INFO
        filter { call -> call.request.path().startsWith("/") }
    }

    routing {
        get("/") {
showTables@
            //       
            call.respondText("HELLO WORLD!", contentType = ContentType.Text.Plain)
        }

        get("/json/gson") {
showPool@            
            //      
            call.respond(mapOf("hello" to "world"))
        }
    }
}

, application.conf ktor.deployment port . , :

        watch = [ /home/votez/cc/ktor-reactive-db/build/classes/kotlin/main/ ]

— , devtools, . , .

initDatabase ( ) .

initDatabase@
    "" // ,        
    val connectionFactory = H2ConnectionFactory(
        H2ConnectionConfiguration.builder()
            .inMemory("users")
            .property(H2ConnectionOption.DB_CLOSE_DELAY, "-1")
            .build()
    )

    val poolConfig = ConnectionPoolConfiguration.builder(connectionFactory)
        .maxIdleTime(10.seconds.toJavaDuration()) //     @ExperimentalTime
        .maxSize(20)
        .build()

    val pool = ConnectionPool(poolConfig)

@ExperimentalTime — API.

, ( ) . , . :

applicationStarted@
    //       
    environment.monitor.subscribe(ApplicationStarted) {
        launch {
            val defer : Mono<Int> = pool.warmup()
            defer.awaitFirst()                             //   
            log.debug("Pool is hot, welcome!") //    
        }
    }

— launch . , , — , . - ! "" (awaitFirst Kotlin), . : Reactor- Mono Flux ? . " -"? , — . — . , , Reactor RxJava . — "!" " !". , , - . — RxJava Reactor — !

, . , .

CRUD , , , H2 : TABLES. - :

data class Tables(    val catalog: String?,    val schema: String?,    val tableName: String?,    val tableType: String?,    val storageType: String?,    val id: String?,    val typeName: String?,    val tableClass: String?,    val rowCountEstimate: Long?) 

typealias , IDE

typealias R2DBCResult = io.r2dbc.spi.Result

. , R2DBC:

get("/tables") {
            showTables@
            //       
            ""
            val connection = pool.create().awaitSingle() //    -  
            val list = try { 
                val result: List<R2DBCResult> = connection.createStatement(
                        """
                            select 
                                TABLE_CATALOG, TABLE_SCHEMA, TABLE_NAME, TABLE_TYPE, STORAGE_TYPE, SQL, ID, TYPE_NAME, TABLE_CLASS, ROW_COUNT_ESTIMATE
                            from 
                                INFORMATION_SCHEMA.TABLES
                                """.trimIndent()
                    )
                    .execute()  //    
                    .toFlux()   //      Reactor Flux
                    .collectList()  //             
                    .awaitFirst()   // ,    -    .
   convertData@
                ""
                TODO() //    ,    
            } finally {
                connection
                    .close()    //   
                    .awaitFirstOrNull() //   null -    .
            }

            call.respond(list)
        }

, Reactor map/flatMap . map/flatMap , , . finally — ( ) doFinally . , README conn.close(), . — , — . , map await, .

awaitFirstOrNull() Mono, onNext() .

, — , , .

R2DBC - (map) . companion object -:

 companion object DB{
        fun ofRow(r: Row, unused: RowMetadata) = Tables(
            r.get("TABLE_CATALOG", String::class.java),
            r.get("TABLE_SCHEMA", String::class.java),
            r.get("TABLE_NAME", String::class.java),
            r.get("TABLE_TYPE", String::class.java),
            r.get("STORAGE_TYPE", String::class.java),
            r.get("ID", Integer::class.java)?.toString(),
            r.get("TYPE_NAME", String::class.java),
            r.get("TABLE_CLASS", String::class.java),
            r.get("ROW_COUNT_ESTIMATE", java.lang.Long::class.java)?.toLong() ?: 0 //     Java ,     
        )
    }

, JDBC .

map Reactor, ReactiveStream Java Stream, R2DBC.

   convertData@
                result.flatMap {//      .  -   ,    
                    it
                        .map(Tables.DB::ofRow)  //     
                        .toFlux()               //      Reactor Flux 
                        .collectList()          //             
                        .awaitFirst()          // ,    -    .
                }

, :

curl http://localhost:8080/tables

Si JSON est moche, il est configuré dans la section gson de l' application ( setPrettyPrinting ).

Nous donnons des statistiques de pool

Eh bien et pour la beauté (puisque nous n'avons pas connecté le module standard de métriques), nous ajouterons un point pour afficher les statistiques du pool. Les moteurs de modèle nous sont superflus, car nos outils de langage nous permettent de:

        get("/pool") {
            showPool@
            //      
            call.respondText {
                (pool.metrics.map {
                    """
                Max allocated size:                 ${it.maxAllocatedSize}
                Max pending size  :                 ${it.maxPendingAcquireSize}
                Acquired size     :                 ${it.acquiredSize()}
                Pending acquire size:               ${it.pendingAcquireSize()}
                Allocated size    :                 ${it.allocatedSize()}
                Idle size         :                 ${it.idleSize()}\n
            """.trimIndent()
                }.orElse("NO METRICS"))
            }
        }

Bien sûr, si vous le souhaitez, vous pouvez le faire via HTML DSL.

résultats

Il est possible de se faire des amis de Coroutines avec des flux réactifs, mais il est nécessaire de basculer entre les styles réactifs et impératifs - de préférence moins souvent et d'essayer d'adhérer à un style.

Pas seulement un seul ressort!

L'accès réactif à la base de données n'est pas aussi beau qu'après le maquillage Spring Data JPA, mais vous pouvez l'utiliser.