Reaktor, WebFlux, Kotlin Coroutines, atau Asynchrony dengan Contoh Sederhana

Banyak layanan di dunia modern, sebagian besar, "tidak melakukan apa-apa". Tugas mereka dikurangi menjadi permintaan untuk database / layanan / cache lain dan agregasi semua data ini sesuai dengan berbagai aturan dan berbagai logika bisnis. Oleh karena itu, tidak mengherankan bahwa bahasa seperti Golang muncul, dengan sistem kompetitif bawaan yang nyaman yang membuatnya mudah untuk mengatur kode non-pemblokiran.

Di dunia JVM, segalanya sedikit lebih rumit. Ada sejumlah besar kerangka kerja dan pustaka yang memblokir utas saat digunakan. Jadi stdlib sendiri dapat melakukan hal yang sama di waktu-waktu tertentu. Dan di Jawa tidak ada mekanisme serupa dengan goroutin di Golang.

Namun demikian, JVM secara aktif berkembang dan peluang baru yang menarik muncul. Ada Kotlin dengan coroutine, yang dalam penggunaannya sangat mirip dengan goroutine Gorang (meskipun mereka diterapkan dengan cara yang sama sekali berbeda). Ada JEP Loom, yang akan membawa serat ke JVM di masa depan. Salah satu kerangka kerja web paling populer - Spring - baru-baru ini menambahkan kemampuan untuk membuat layanan yang sepenuhnya non-blocking di Webflux. Dan dengan rilis terbaru dari boot Spring 2.2, integrasi dengan Kotlin bahkan lebih baik.

Saya mengusulkan, menggunakan contoh layanan kecil untuk mentransfer uang dari satu kartu ke kartu lain, untuk menulis aplikasi pada boot Spring 2.2 dan Kotlin untuk integrasi dengan beberapa layanan eksternal.

Baik jika Anda sudah terbiasa dengan Java, Kotlin, Gradle, Spring, Spring boot 2, Reactor, fluks Web , Tomcat, Netty, Kotlin outoutines, Gradle Kotlin DSL atau bahkan memiliki gelar Ph.D. Tetapi jika tidak, itu tidak masalah. Kode ini akan disederhanakan secara maksimal, dan bahkan jika Anda bukan dari dunia JVM, saya harap semuanya akan jelas bagi Anda.

Jika Anda berencana untuk menulis layanan sendiri, pastikan semua yang Anda butuhkan sudah diinstal:

• Java 8+
• Docker dan Docker Compose;
• CURL dan lebih disukai jq ;
• Git
• lebih disukai IDE untuk Kotlin (Intellij Idea, Eclipse, VS, vim , dll.). Tapi itu mungkin di notebook.

Contoh akan berisi kosong untuk implementasi dalam layanan, dan implementasi yang sudah tertulis. Pertama, jalankan instalasi dan perakitan dan perhatikan lebih dekat layanan dan API mereka.

Contoh layanan dan API itu sendiri dibuat hanya untuk tujuan ilustrasi, jangan transfer semua AS IS ke prod Anda!

Pertama, kita mengkloning repositori dengan layanan ke diri kita sendiri, integrasi yang dengannya kita akan melakukannya, dan pergi ke direktori:

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

Di terminal terpisah, kami mengumpulkan semua aplikasi menggunakan gradle , di mana setelah pembangunan yang berhasil semua layanan akan diluncurkan menggunakan docker-compose .

 ./gradlew build && docker-compose up 

Sementara semuanya diunduh dan diinstal, pertimbangkan proyek dengan layanan.

Permintaan dengan token, nomor kartu untuk transfer, dan jumlah yang akan ditransfer antar kartu akan diterima di pintu masuk layanan (layanan Demo):

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

authToken token authToken , authToken harus pergi ke layanan AUTH dan mendapatkan userId , yang dengannya Anda kemudian dapat membuat permintaan ke USER dan mengeluarkan semua informasi tambahan tentang pengguna. AUTH juga akan mengembalikan informasi tentang mana dari tiga layanan yang dapat kita akses. Contoh tanggapan dari AUTH :

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

Untuk mentransfer antar kartu, pertama pergi dengan setiap nomor kartu dalam CARD . Menanggapi permintaan, kami akan menerima cardId , kemudian dengan itu kami mengirim permintaan ke PAYMENT dan melakukan transfer. Dan yang terakhir - sekali lagi kami mengirim permintaan ke PAYMENT dengan dari fromCardId dan mencari tahu saldo saat ini.

Untuk meniru sedikit keterlambatan dalam layanan, nilai variabel lingkungan TIMEOUT dilemparkan ke semua wadah, di mana penundaan respons diatur dalam milidetik. Dan untuk mendiversifikasi respons dari AUTH , dimungkinkan untuk memvariasikan nilai SUCCESS_RATE , yang mengontrol kemungkinan respons true untuk layanan.

File 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" 

Untuk semua layanan, port forwarding dari 8081 ke 8084 dilakukan untuk dengan mudah menjangkau mereka secara langsung.

Mari beralih ke penulisan Demo service . Pertama, mari kita coba untuk menulis implementasi yang seringkuh mungkin, tanpa asinkron dan konkurensi. Untuk melakukan ini, ambil boot Spring 2.2.1, Kotlin dan kosong untuk layanan. Kami mengkloning repositori dan pergi ke cabang spring-mvc-start :

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

Buka file demo.Controller . Ini memiliki satu-satunya metode processRequest kosong yang implementasi harus ditulis.

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

Permintaan untuk transfer antar kartu akan diterima di pintu masuk metode.

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

Untuk membuatnya lebih mudah, semua metode untuk integrasi dengan layanan lain telah dibuat, Anda hanya perlu menghubungkannya dengan benar. Hanya ada lima metode, satu untuk setiap tindakan. Panggilan ke layanan lain sendiri diterapkan pada panggilan pemblokiran Spring RestTemplate .

Metode contoh untuk memanggil 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'") } 

Mari kita beralih ke implementasi metode. Komentar menunjukkan prosedur dan respons apa yang diharapkan pada output:

  @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 = // ) } 

Pertama, kami menerapkan metode sesederhana mungkin, tanpa memperhitungkan bahwa AUTH dapat menolak akses kami ke layanan lain. Coba lakukan sendiri. Ketika ternyata (atau setelah beralih ke cabang spring-mvc ), Anda dapat memeriksa pengoperasian layanan sebagai berikut:

 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 ) } 

Mulai layanan (dari folder layanan demo):

 ./gradlew bootRun 

Kami mengirim permintaan ke titik akhir:

 ./demo-request.sh 

Sebagai tanggapan, kami mendapatkan sesuatu seperti ini:

 ➜ 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 } 

Secara total, Anda perlu membuat 6 permintaan untuk mengimplementasikan layanan. Dan mengingat bahwa masing-masing dari mereka merespons dengan penundaan 100 ms, total waktu tidak boleh kurang dari 600 ms. Pada kenyataannya, ternyata sekitar 700 ms, dengan mempertimbangkan semua overhead. Sejauh ini kodenya cukup sederhana, dan jika kita sekarang ingin menambahkan cek respons AUTH untuk mengakses layanan lain, ini tidak akan sulit dilakukan (seperti halnya refactoring lainnya).

Tapi mari kita pikirkan bagaimana Anda dapat mempercepat eksekusi permintaan. Jika Anda tidak memperhitungkan verifikasi jawaban dari AUTH , maka kami memiliki 2 tugas independen:

• mendapatkan userId dan meminta data dari USER ;
• menerima cardId untuk setiap kartu, melakukan pembayaran dan menerima jumlah total.

Tugas-tugas ini dapat dilakukan secara independen satu sama lain. Maka total waktu eksekusi akan tergantung pada rantai panggilan terpanjang (dalam hal ini, yang kedua) dan akan dieksekusi secara total untuk 300 ms + X ms overhead.

Mengingat bahwa panggilan itu sendiri memblokir, satu-satunya cara untuk menjalankan permintaan paralel adalah dengan menjalankannya di utas terpisah. Anda dapat membuat Utas terpisah untuk setiap panggilan, tetapi akan sangat mahal. Cara lain adalah menjalankan tugas di ThreadPool. Sepintas, solusi semacam itu terlihat tepat, dan waktu akan benar-benar berkurang. Misalnya, kita dapat menjalankan kueri di CompletableFuture. Ini memungkinkan Anda untuk menjalankan tugas latar belakang dengan memanggil metode dengan postfix async . Dan jika Anda tidak menentukan ThreadPool tertentu saat memanggil metode, tugas akan diluncurkan di ForkJoinPool.commonPool() . Cobalah untuk menulis implementasi sendiri atau buka cabang 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 ) } 

Jika sekarang kita mengukur waktu permintaan, itu akan berada di wilayah 360 ms. Dibandingkan dengan versi aslinya, total waktu berkurang hampir 2 kali lipat. Kode itu sendiri menjadi sedikit lebih rumit, tetapi sejauh ini masih tidak sulit untuk memodifikasinya. Dan jika di sini kita ingin menambahkan cek respons dari AUTH , maka ini tidak sulit.

Tetapi bagaimana jika kita memiliki sejumlah besar permintaan masuk untuk layanan itu sendiri? Katakan sekitar 1000 permintaan simultan? Dengan pendekatan ini, ternyata cukup cepat bahwa semua utas ThreadPool sibuk membuat panggilan pemblokiran. Dan kita sampai pada kesimpulan bahwa versi saat ini juga tidak sesuai.

Tetap hanya melakukan sesuatu dengan panggilan layanan sendiri. Anda dapat memodifikasi kueri dan membuatnya non-pemblokiran. Kemudian metode untuk memanggil layanan akan mengembalikan CompletableFuture, Flux, Observable, Deferred, Promise atau objek serupa yang digunakan untuk membangun rantai harapan. Dengan pendekatan ini, kita tidak perlu melakukan panggilan pada aliran yang terpisah - itu akan cukup untuk memilikinya (atau setidaknya kumpulan aliran yang terpisah kecil) yang telah kita pinjam untuk memproses permintaan.

Bisakah kita menahan beban berat pada layanan ini? Untuk menjawab pertanyaan ini, perhatikan Tomcat, yang digunakan pada Spring boot 2.2.1 di starter org.springframework.boot:spring-boot-starter-web . Itu dibangun sehingga utas dari ThreadPool dialokasikan untuk setiap permintaan masuk untuk pemrosesan. Dan dengan tidak adanya arus bebas, permintaan baru akan menjadi "antrian" menunggu. Tetapi layanan kami sendiri hanya mengirimkan permintaan ke layanan lain. Mengalokasikan seluruh aliran di bawahnya dan memblokirnya sampai jawaban dari semua orang datang, terlihat, untuk membuatnya lebih halus, berlebihan.

Untungnya, Spring baru-baru ini memungkinkan untuk menggunakan server web non-pemblokiran berdasarkan Netty atau Undertow. Untuk melakukan ini, Anda hanya perlu mengubah spring-boot-starter-web menjadi spring-boot-starter-webflux dan sedikit mengubah metode untuk memproses permintaan di mana permintaan dan respons akan "dibungkus" dalam Mono. Ini disebabkan oleh fakta bahwa Webflux dibangun atas dasar Reaktor, dan karena itu sekarang dalam metode yang Anda perlukan untuk membangun rantai transformasi Mono.

Cobalah menulis implementasi metode Anda yang tidak menghalangi. Untuk melakukannya, buka cabang spring-webflux-start . Harap perhatikan bahwa starter untuk Spring Boot telah berubah, di mana versi dengan Webflux sekarang digunakan, dan implementasi permintaan ke layanan lain yang telah ditulis ulang untuk menggunakan WebClient non-blocking juga telah berubah.

Metode contoh untuk memanggil AUTH:

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

Implementasi dari contoh pertama dimasukkan ke dalam isi metode processRequest dalam komentar. Cobalah untuk menulis ulang sendiri di Reactor. Seperti terakhir kali, pertama-tama buat versi tanpa memperhitungkan cek dari AUTH , dan kemudian lihat betapa sulitnya menambahkannya:

 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() } 

Setelah berurusan dengan ini, Anda dapat membandingkan dengan implementasi saya dari cabang 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 ) } } 

Setuju bahwa sekarang menulis implementasi (dibandingkan dengan pendekatan pemblokiran sebelumnya) menjadi lebih sulit. Dan jika kita ingin menambahkan "lupa" cek dari AUTH , maka ini tidak akan mudah dilakukan.

Ini adalah inti dari pendekatan reaktif. Ini bagus untuk membangun rantai pemrosesan yang tidak bercabang. Tetapi jika percabangan muncul, maka kodenya tidak lagi begitu sederhana.

Kotlin coroutine, yang sangat ramah dengan kode asinkron / reaktif, dapat membantu di sini. Selain itu, ada sejumlah besar pembungkus tertulis untuk Reactor , CompletableFuture , dll. Tetapi bahkan jika Anda tidak menemukan yang tepat, Anda selalu dapat menulis sendiri, menggunakan pembangun khusus.

Mari kita menulis ulang implementasi pada coroutine sendiri. Untuk melakukannya, buka cabang spring-webflux-coroutines-start . Ketergantungan yang diperlukan ditambahkan padanya di 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") 

Dan metode processRequest berubah processRequest :

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

Ini tidak lagi membutuhkan Mono dan diterjemahkan hanya menjadi fungsi menangguhkan (berkat integrasi Spring dan Kotlin). Menimbang bahwa kita akan membuat coroutine tambahan dalam metode ini, kita perlu membuat scout coroutineScope anak (untuk pemahaman tentang alasan-alasan untuk menciptakan ruang lingkup tambahan, lihat posting Roman Elizarov tentang Structured concurrency ). Harap perhatikan bahwa panggilan layanan lain tidak berubah sama sekali. Mereka mengembalikan Mono yang sama di mana metode suspend awaitFirst dapat dipanggil untuk "menunggu" untuk hasil query.

Jika coroutine masih merupakan konsep baru untuk Anda, maka ada panduan luar biasa dengan deskripsi terperinci. Coba tulis implementasi Anda sendiri dari metode processRequest atau buka cabang 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 ) } 

Anda dapat membandingkan kode dengan pendekatan reaktif. Dengan coroutine, Anda tidak perlu memikirkan semua poin cabang terlebih dahulu. Kami hanya dapat memanggil metode await dan bercabang tugas asinkron di async di tempat yang tepat. Kode tetap sama mungkin dengan versi langsung yang asli, yang sama sekali tidak sulit untuk diubah. Dan faktor penting adalah bahwa coroutine hanya tertanam dalam kode reaktif.

Anda bahkan mungkin lebih menyukai pendekatan reaktif untuk tugas ini, tetapi banyak orang yang disurvei merasa lebih sulit. Secara umum, kedua pendekatan menyelesaikan masalah mereka dan Anda dapat menggunakan salah satu yang Anda sukai. Ngomong-ngomong, baru-baru ini di Kotlin ada juga kesempatan untuk membuat coroutine “dingin” dengan Flow, yang sangat mirip dengan Reactor. Benar, mereka masih dalam tahap percobaan, tetapi sekarang Anda dapat melihat implementasi saat ini dan mencobanya dalam kode Anda.

Saya ingin mengakhiri di sini dan akhirnya meninggalkan tautan yang bermanfaat:

• Panduan Coroutine
• Reaktor Coroutines
• Coroutines CompletableFuture
• Roman Elizarov tentang coroutine
• Kotlin Mengalir dan Coroutine
• Konkurensi terstruktur
• Boot musim semi mencintai Kotlin

Saya harap Anda tertarik dan Anda berhasil menulis implementasi metode untuk semua metode sendiri. Dan, tentu saja, saya ingin percaya bahwa Anda menyukai opsi dengan coroutine lebih =)

Terima kasih kepada semua orang yang membaca sampai akhir!