Reactor, WebFlux, Kotlin Coroutines ou Asynchrony avec un exemple simple

De nombreux services dans le monde moderne, pour la plupart, «ne font rien». Leurs tâches sont réduites aux demandes d'autres bases de données / services / caches et à l'agrégation de toutes ces données selon différentes règles et différentes logiques métier. Par conséquent, il n'est pas surprenant que des langues telles que Golang apparaissent, avec un système compétitif intégré pratique qui facilite l'organisation du code non bloquant.

Dans le monde JVM, les choses sont un peu plus compliquées. Il existe un grand nombre de frameworks et de bibliothèques qui bloquent les threads lorsqu'ils sont utilisés. Donc, stdlib lui-même peut parfois faire la même chose. Et à Java, il n'y a pas de mécanisme similaire à celui des goroutins à Golang.

Néanmoins, la JVM se développe activement et de nouvelles opportunités intéressantes apparaissent. Il y a Kotlin avec des coroutines, qui dans leur utilisation sont très similaires aux goroutines Gorang (bien qu'elles soient implémentées d'une manière complètement différente). Il y a JEP Loom, qui apportera des fibres à la JVM à l'avenir. L'un des frameworks Web les plus populaires - Spring - a récemment ajouté la possibilité de créer des services totalement non bloquants sur Webflux. Et avec la récente version de Spring boot 2.2, l'intégration avec Kotlin est encore meilleure.

Je propose, en utilisant l'exemple d'un petit service de transfert d'argent d'une carte à une autre, d'écrire une application sur Spring boot 2.2 et Kotlin pour intégration avec plusieurs services externes.

C'est bien si vous connaissez déjà Java, Kotlin, Gradle, Spring, Spring boot 2, Reactor, Web flux, Tomcat, Netty, Kotlin oroutines, Gradle Kotlin DSL ou même si vous avez un doctorat. Mais sinon, cela n'a pas d'importance. Le code sera simplifié au maximum, et même si vous n'êtes pas du monde JVM, j'espère que vous comprendrez tout.

Si vous prévoyez d'écrire un service vous-même, assurez-vous que tout ce dont vous avez besoin est installé:

• Java 8+
• Docker et Docker Compose;
• cURL et de préférence jq ;
• Git
• de préférence un IDE pour Kotlin (Intellij Idea, Eclipse, VS, vim , etc.). Mais c'est possible dans un cahier.

Les exemples contiendront à la fois des blancs pour l'implémentation dans le service et une implémentation déjà écrite. Tout d'abord, exécutez l'installation et l'assemblage et examinez de plus près les services et leurs API.

L'exemple des services et de l'API lui-même est fourni à des fins d'illustration uniquement; ne transférez pas tous AS IS à votre produit!

Tout d'abord, nous clonons le référentiel avec des services pour nous-mêmes, l'intégration avec laquelle nous le ferons, et allons dans le répertoire:

 git clone https://github.com/evgzakharov/spring-demo-services && cd spring-demo-services 

Dans un terminal séparé, nous collectons toutes les applications en utilisant gradle , où après une construction réussie, tous les services seront lancés en utilisant docker-compose .

 ./gradlew build && docker-compose up 

Pendant que tout est téléchargé et installé, envisagez un projet avec des services.

Une demande avec un token, des numéros de carte à transférer et le montant à transférer entre les cartes sera reçue à l'entrée du service (service de démonstration):

 { "authToken": "auth-token1", "cardFrom": "55593478", "cardTo": "55592020", "amount": "10.1" } 

Selon le jeton authToken , authToken devez accéder au service AUTH et obtenir userId , avec lequel vous pouvez ensuite faire une demande à USER et extraire toutes les informations supplémentaires sur l'utilisateur. AUTH retournera également des informations sur lequel des trois services nous pouvons accéder. Exemple de réponse de AUTH :

 { "userId": 158, "cardAccess": true, "paymentAccess": true, "userAccess": true } 

Pour transférer entre les cartes, allez d'abord avec chaque numéro de carte dans la CARD . En réponse aux demandes, nous recevrons cardId , puis avec eux nous envoyons une demande au PAYMENT et effectuons un virement. Et le dernier - une fois de plus, nous envoyons une demande au PAYMENT avec fromCardId et découvrons le solde actuel.

Pour émuler un petit retard dans les services, la valeur de la variable d'environnement TIMEOUT est lancée dans tous les conteneurs, dans lesquels le délai de réponse est défini en millisecondes. Et pour diversifier les réponses de AUTH , il est possible de faire varier la valeur de SUCCESS_RATE , qui contrôle la probabilité d'une true réponse pour le service.

Fichier Docker-compose.yaml:

 version: '3' services: service-auth: build: service-auth image: service-auth:1.0.0 environment: - SUCCESS_RATE=1.0 - TIMEOUT=100 ports: - "8081:8080" service-card: build: service-card image: service-card:1.0.0 environment: - TIMEOUT=100 ports: - "8082:8080" service-payment: build: service-payment image: service-payment:1.0.0 environment: - TIMEOUT=100 ports: - "8083:8080" service-user: build: service-user image: service-user:1.0.0 environment: - TIMEOUT=100 ports: - "8084:8080" 

Pour tous les services, la redirection de port de 8081 à 8084 est effectuée pour les atteindre directement directement.

Passons maintenant à l'écriture du Demo service . Tout d'abord, essayons d'écrire l'implémentation aussi maladroite que possible, sans asynchronie et simultanéité. Pour ce faire, prenez Spring boot 2.2.1, Kotlin et un blanc pour le service. Nous clonons le référentiel et passons à la branche spring-mvc-start :

 git clone https://github.com/evgzakharov/demo-service && cd demo-service && git checkout spring-mvc-start 

Accédez au fichier demo.Controller . Il possède la seule méthode processRequest vide pour laquelle une implémentation doit être écrite.

  @PostMapping fun processRequest(@RequestBody serviceRequest: ServiceRequest): Response { .. } 

Une demande de virement entre cartes sera reçue à l'entrée de la méthode.

 data class ServiceRequest( val authToken: String, val cardFrom: String, val cardTo: String, val amount: BigDecimal ) 

Pour vous faciliter la tâche, toutes les méthodes d'intégration avec d'autres services ont déjà été réalisées, il vous suffit de les connecter correctement. Il n'y a que cinq méthodes, une pour chaque action. Les appels vers d'autres services eux-mêmes sont implémentés sur l'appel de blocage Spring RestTemplate .

Exemple de méthode pour appeler AUTH :

 private fun getAuthInfo(token: String): AuthInfo { log.info("getAuthInfo") return restTemplate.getForEntity("${demoConfig.auth}/{token}", AuthInfo::class.java, token) .body ?: throw RuntimeException("couldn't find user by token='$token'") } 

Passons à l'implémentation de la méthode. Les commentaires indiquent la procédure et la réponse attendue en sortie:

  @PostMapping fun processRequest(@RequestBody serviceRequest: ServiceRequest): Response { //1) get auth info from service by token -> userId //2) find user info by userId from 1. //3) 4) find cards info for each card in serviceRequest // 5) make transaction for known cards by calling sendMoney(id1, id2, amount) // 6) after payment get payment info by fromCardId TODO("return SuccessResponse") // SuccessResponse( // amount = , // userName = , // userSurname = , // userAge = // ) } 

Tout d'abord, nous mettons en œuvre la méthode le plus simplement possible, sans tenir compte du fait AUTH peut nous refuser l'accès à d'autres services. Essayez de le faire vous-même. Lorsqu'il s'avère (ou après le passage à la branche spring-mvc ), vous pouvez vérifier le fonctionnement du service comme suit:

 fun processRequest(@RequestBody serviceRequest: ServiceRequest): Response { val authInfo = getAuthInfo(serviceRequest.authToken) val userInfo = findUser(authInfo.userId) val cardFromInfo = findCardInfo(serviceRequest.cardFrom) val cardToInfo = findCardInfo(serviceRequest.cardTo) sendMoney(cardFromInfo.cardId, cardToInfo.cardId, serviceRequest.amount) val paymentInfo = getPaymentInfo(cardFromInfo.cardId) return SuccessResponse( amount = paymentInfo.currentAmount, userName = userInfo.name, userSurname = userInfo.surname, userAge = userInfo.age ) } 

Démarrez le service (à partir du dossier du service de démonstration):

 ./gradlew bootRun 

Nous envoyons une demande au point final:

 ./demo-request.sh 

En réponse, nous obtenons quelque chose comme ceci:

 ➜ demo-service git:(spring-mvc) ✗ ./demo-request.sh + curl -XPOST http://localhost:8080/ -d @demo-payment-request.json -H 'Content-Type: application/json; charset=UTF-8' + jq . % Total % Received % Xferd Average Speed Time Time Time Current Dload Upload Total Spent Left Speed 100 182 0 85 100 97 20 23 0:00:04 0:00:04 --:--:-- 23 { "amount": 989.9, "userName": "Vasia", "userSurname": "Pupkin", "userAge": 18, "status": true } 

Au total, vous devez effectuer 6 demandes pour mettre en œuvre le service. Et étant donné que chacun d'eux répond avec un retard de 100 ms, le temps total ne peut pas être inférieur à 600 ms. En réalité, il s'avère environ 700 ms, en tenant compte de tous les frais généraux. Jusqu'à présent, le code est assez simple, et si nous voulons maintenant ajouter une vérification de réponse AUTH pour accéder à d'autres services, cela ne sera pas difficile à faire (comme tout autre refactoring).

Mais réfléchissons à la façon dont vous pouvez accélérer l'exécution des requêtes. Si vous ne prenez pas en compte la vérification de la réponse de AUTH , alors nous avons 2 tâches indépendantes:

• obtenir l' userId et demander des données à l' USER ;
• recevoir cardId pour chaque carte, effectuer un paiement et recevoir le montant total.

Ces tâches peuvent être effectuées indépendamment les unes des autres. Ensuite, le temps d'exécution total dépendra de la plus longue chaîne d'appels (dans ce cas, le second) et sera exécuté au total pendant 300 ms + X ms de surcharge.

Étant donné que les appels eux-mêmes bloquent, la seule façon d'exécuter des requêtes parallèles est de les exécuter sur des threads séparés. Vous pouvez créer un thread séparé pour chaque appel, mais cela coûtera très cher. Une autre façon consiste à exécuter des tâches sur ThreadPool. À première vue, une telle solution semble appropriée et le temps va vraiment diminuer. Par exemple, nous pouvons exécuter des requêtes sur CompletableFuture. Il vous permet d'exécuter des tâches en arrière-plan en appelant des méthodes avec le suffixe async . Et si vous ne spécifiez pas de ThreadPool spécifique lors de l'appel de méthodes, les tâches seront lancées sur ForkJoinPool.commonPool() . Essayez d'écrire une implémentation vous-même ou accédez à la branche spring-mvc-async .

 fun processRequest(@RequestBody serviceRequest: ServiceRequest): Response { val authInfoFuture = CompletableFuture.supplyAsync { getAuthInfo(serviceRequest.authToken) } val userInfoFuture = authInfoFuture.thenApplyAsync { findUser(it.userId) } val cardFromInfo = CompletableFuture.supplyAsync { findCardInfo(serviceRequest.cardFrom) } val cardToInfo = CompletableFuture.supplyAsync { findCardInfo(serviceRequest.cardTo) } val waitAll = CompletableFuture.allOf(cardFromInfo, cardToInfo) val paymentInfoFuture = waitAll .thenApplyAsync { sendMoney(cardFromInfo.get().cardId, cardToInfo.get().cardId, serviceRequest.amount) } .thenApplyAsync { getPaymentInfo(cardFromInfo.get().cardId) } val paymentInfo = paymentInfoFuture.get() val userInfo = userInfoFuture.get() log.info("result") return SuccessResponse( amount = paymentInfo.currentAmount, userName = userInfo.name, userSurname = userInfo.surname, userAge = userInfo.age ) } 

Si nous mesurons maintenant le temps de requête, il sera de l'ordre de 360 ​​ms. Par rapport à la version originale, le temps total a diminué de près de 2 fois. Le code lui-même est devenu un peu plus compliqué, mais jusqu'à présent, il n'est toujours pas difficile de le modifier. Et si ici nous voulons ajouter un contrôle de réponse d' AUTH , alors ce n'est pas difficile.

Mais que faire si nous avons un grand nombre de demandes entrantes pour le service lui-même? Dites environ 1000 demandes simultanées? Avec cette approche, il s'avère assez rapidement que tous les threads ThreadPool sont occupés à effectuer des appels de blocage. Et nous arrivons à la conclusion que la version actuelle ne convient pas non plus.

Il ne reste plus qu'à faire quelque chose avec les appels de service eux-mêmes. Vous pouvez modifier les requêtes et les rendre non bloquantes. Ensuite, les méthodes d'appel des services renverront CompletableFuture, Flux, Observable, Deferred, Promise ou un objet similaire sur lequel construire une chaîne d'attentes. Avec cette approche, nous n'avons pas besoin de faire des appels sur des flux séparés - il suffira d'en avoir un (ou au moins un petit pool séparé de flux) que nous avons déjà emprunté pour traiter les demandes.

Pouvons-nous maintenant supporter la lourde charge du service? Pour répondre à cette question, examinez de près Tomcat, qui est utilisé dans Spring boot 2.2.1 dans le starter org.springframework.boot:spring-boot-starter-web . Il est construit de sorte qu'un thread de ThreadPool est alloué pour chaque demande entrante pour son traitement. Et en l'absence de flux libres, les nouvelles demandes deviendront une «file d'attente» d'attente. Mais notre service lui-même envoie uniquement des demandes à d'autres services. Allouer un flux entier en dessous et le bloquer jusqu'à ce que les réponses de tout le monde arrivent, semble, pour le dire doucement, superflu.

Heureusement, Spring a récemment permis d'utiliser un serveur Web non bloquant basé sur Netty ou Undertow. Pour ce faire, il vous suffit de remplacer le spring-boot-starter-web spring-boot-starter-webflux et de modifier légèrement la méthode de traitement des demandes dans laquelle la demande et la réponse seront "encapsulées" en Mono. Cela est dû au fait que Webflux est construit sur la base de Reactor, et donc maintenant dans la méthode dont vous avez besoin pour construire une chaîne de transformations mono.

Essayez d'écrire votre propre implémentation non bloquante de la méthode. Pour ce faire, accédez à la branche spring-webflux-start . Veuillez noter que le démarreur de Spring Boot a changé, où la version avec Webflux est maintenant utilisée, et l'implémentation des demandes à d'autres services qui ont été réécrites pour utiliser WebClient non bloquant a également changé.

Exemple de méthode pour appeler AUTH:

 private fun getAuthInfo(token: String): Mono<AuthInfo> { log.info("getAuthInfo") return WebClient.create().get() .uri("${demoConfig.auth}/$token") .retrieve() .bodyToMono(AuthInfo::class.java) } 

L'implémentation du premier exemple est insérée dans le contenu de la méthode processRequest dans un commentaire. Essayez de le réécrire vous-même sur Reactor. Comme la dernière fois, faites d'abord la version sans prendre en compte les chèques d' AUTH , puis voyez comme il est difficile de les ajouter:

 fun processRequest(@RequestBody serviceRequest: Mono<ServiceRequest>): Mono<Response> { // val authInfo = getAuthInfo(serviceRequest.authToken) // // val userInfo = findUser(authInfo.userId) // // val cardFromInfo = findCardInfo(serviceRequest.cardFrom) // val cardToInfo = findCardInfo(serviceRequest.cardTo) // // sendMoney(cardFromInfo.cardId, cardToInfo.cardId, serviceRequest.amount) // // val paymentInfo = getPaymentInfo(cardFromInfo.cardId) // // log.info("result") // // return SuccessResponse( // amount = paymentInfo.currentAmount, // userName = userInfo.name, // userSurname = userInfo.surname, // userAge = userInfo.age // ) TODO() } 

Après avoir traité cela, vous pouvez comparer avec mon implémentation de la spring-webflux :

 fun processRequest(@RequestBody serviceRequest: Mono<ServiceRequest>): Mono<Response> { val cacheRequest = serviceRequest.cache() val userInfoMono = cacheRequest.flatMap { getAuthInfo(it.authToken) }.flatMap { findUser(it.userId) } val cardFromInfoMono = cacheRequest.flatMap { findCardInfo(it.cardFrom) } val cardToInfoMono = cacheRequest.flatMap { findCardInfo(it.cardTo) } val paymentInfoMono = cardFromInfoMono.zipWith(cardToInfoMono) .flatMap { (cardFromInfo, cardToInfo) -> cacheRequest.flatMap { request -> sendMoney(cardFromInfo.cardId, cardToInfo.cardId, request.amount).map { cardFromInfo } } }.flatMap { getPaymentInfo(it.cardId) } return userInfoMono.zipWith(paymentInfoMono) .map { (userInfo, paymentInfo) -> log.info("result") SuccessResponse( amount = paymentInfo.currentAmount, userName = userInfo.name, userSurname = userInfo.surname, userAge = userInfo.age ) } } 

Convenez que l'écriture d'une implémentation (par rapport à l'approche de blocage précédente) est devenue plus difficile. Et si nous voulons ajouter des chèques «oubliés» d' AUTH , ce ne sera pas si facile à faire.

C'est l'essence même de l'approche réactive. Il est idéal pour construire des chaînes de transformation non ramifiées. Mais si la ramification apparaît, le code n'est plus aussi simple.

Les coroutines Kotlin, qui sont très amicales avec n'importe quel code asynchrone / réactif, peuvent vous aider ici. De plus, il existe un grand nombre de wrappers écrits pour Reactor , CompletableFuture , etc. Mais même si vous ne trouvez pas le bon, vous pouvez toujours l'écrire vous-même, en utilisant des constructeurs spéciaux.

Réécrivons nous-mêmes l'implémentation sur les coroutines. Pour ce faire, accédez à la branche spring-webflux-coroutines-start . Les dépendances nécessaires y sont ajoutées dans build.gradle.kts:

 implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:$kotlinCoroutinesVersion") implementation("org.jetbrains.kotlinx:kotlinx-coroutines-reactive:$kotlinCoroutinesVersion") implementation("org.jetbrains.kotlinx:kotlinx-coroutines-reactor:$kotlinCoroutinesVersion") 

Et la méthode processRequest change un processRequest :

 suspend fun processRequest(@RequestBody serviceRequest: ServiceRequest): Response = coroutineScope { //TODO() } 

Il n'a plus besoin de Mono et se traduit simplement en une fonction de suspension (grâce à l'intégration de Spring et Kotlin). Étant donné que nous allons créer des coroutines supplémentaires dans la méthode, nous devrons créer un coroutineScope scout enfant (pour comprendre les raisons de la création d'une étendue supplémentaire, voir le post de Roman Elizarov sur la concurrence structurée ). Veuillez noter que les autres appels de service n'ont pas changé du tout. Ils renvoient le même Mono sur lequel la méthode de suspend waitFirst peut être appelée pour «attendre» le résultat de la requête.

Si les coroutines sont encore un nouveau concept pour vous, alors il y a un merveilleux guide avec une description détaillée. Essayez d'écrire votre propre implémentation de la méthode processRequest ou accédez à la branche spring-webflux-coroutines :

 suspend fun processRequest(@RequestBody serviceRequest: ServiceRequest): Response = coroutineScope { log.info("start") val userInfoDeferred = async { val authInfo = getAuthInfo(serviceRequest.authToken).awaitFirst() findUser(authInfo.userId).awaitFirst() } val paymentInfoDeferred = async { val cardFromInfoDeferred = async { findCardInfo(serviceRequest.cardFrom).awaitFirst() } val cardToInfoDeferred = async { findCardInfo(serviceRequest.cardTo).awaitFirst() } val cardFromInfo = cardFromInfoDeferred.await() sendMoney(cardFromInfo.cardId, cardToInfoDeferred.await().cardId, serviceRequest.amount).awaitFirst() getPaymentInfo(cardFromInfo.cardId).awaitFirst() } val userInfo = userInfoDeferred.await() val paymentInfo = paymentInfoDeferred.await() log.info("result") SuccessResponse( amount = paymentInfo.currentAmount, userName = userInfo.name, userSurname = userInfo.surname, userAge = userInfo.age ) } 

Vous pouvez comparer le code avec l'approche réactive. Avec les coroutines, vous n'avez pas besoin de réfléchir à l'avance à tous les points de branchement. Nous pouvons simplement appeler les méthodes d' await et dériver les tâches asynchrones dans async aux bons endroits. Le code reste aussi similaire que possible à la version originale simple, qui n'est pas du tout difficile à modifier. Et un facteur important est que les coroutines sont simplement intégrées dans du code réactif.

Vous aimerez peut-être même davantage l'approche réactive pour cette tâche, mais la plupart des personnes interrogées la trouvent plus difficile. En général, les deux approches résolvent leur problème et vous pouvez utiliser celle que vous aimez. Soit dit en passant, récemment à Kotlin, il est également possible de créer des coroutines «froides» avec Flow, qui sont très similaires à Reactor. Certes, ils sont encore au stade expérimental, mais maintenant vous pouvez regarder l'implémentation actuelle et l'essayer dans votre code.

Je veux terminer ici et enfin laisser des liens utiles:

• Guide de voyage Coroutine
• Réacteur Coroutines
• Coroutines CompletableFuture
• Roman Elizarov sur les coroutines
• Flux de Kotlin et coroutines
• Concurrence structurée
• La botte de printemps aime Kotlin

J'espère que vous étiez intéressé et que vous avez réussi à écrire vous-même une implémentation de la méthode pour toutes les méthodes. Et, bien sûr, je veux croire que vous aimez plus l'option avec les coroutines =)

Merci à tous ceux qui ont lu jusqu'au bout!