مفاعل ، WebFlux ، Kotlin Coroutines ، أو غير متزامن مع مثال بسيط

العديد من الخدمات في العالم الحديث ، في معظمها ، "لا تفعل شيئًا". يتم تقليل مهامهم إلى طلبات قواعد البيانات / الخدمات / ذاكرات التخزين المؤقت الأخرى وتجميع كل هذه البيانات وفقًا للقواعد المختلفة ومنطق الأعمال المختلفة. لذلك ، فليس من المستغرب أن تظهر لغات مثل Golang ، مع نظام تنافسي مدمج مناسب يجعل من السهل تنظيم التعليمات البرمجية غير المحظورة.

في عالم JVM ، الأمور معقدة بعض الشيء. هناك عدد كبير من الأطر والمكتبات التي تحظر المواضيع عند استخدامها. حتى stdlib نفسها يمكن أن تفعل الشيء نفسه في بعض الأحيان. وفي جاوة ، لا توجد آلية مماثلة تشبه goroutines في Golang.

ومع ذلك ، JVM تتطور بنشاط وتظهر فرص مثيرة للاهتمام جديدة. هناك Kotlin مع coroutines ، والتي هي في استخدامها تشبه إلى حد كبير goroutines Gorang (على الرغم من أنها تنفذ بطريقة مختلفة تماما). هناك JEP Loom ، والتي ستجلب الألياف إلى JVM في المستقبل. أضاف أحد أكثر أطر الويب شعبية - Spring - مؤخرًا القدرة على إنشاء خدمات غير محظورة تمامًا على Webflux. ومع الإصدار الأخير من Spring boot 2.2 ، أصبح التكامل مع Kotlin أفضل.

أقترح ، باستخدام مثال خدمة صغيرة لتحويل الأموال من بطاقة إلى أخرى ، كتابة طلب على Spring boot 2.2 و Kotlin للتكامل مع العديد من الخدمات الخارجية.

من الجيد أن تكون على دراية بالفعل بالجافا ، Kotlin ، Gradle ، Spring ، Spring boot 2 ، Reactor ، Web flux ، Tomcat ، Netty ، Kotlin oroutines ، Gradle Kotlin DSL أو حتى تحصل على درجة الدكتوراه. لكن إذا لم يكن الأمر كذلك ، فهذا لا يهم. سيتم تبسيط الرمز إلى أقصى حد ، وحتى لو لم تكن من عالم JVM ، آمل أن يكون كل شيء واضحًا لك.

إذا كنت تخطط لكتابة الخدمة بنفسك ، فتأكد من تثبيت كل ما تحتاجه:

• جافا 8+
• عامل الميناء و عامل الميناء يؤلف؛
• حليقة ويفضل jq .
• بوابة.
• ويفضل أن يكون IDE لـ Kotlin (Intellij Idea ، Eclipse ، VS ، vim ، إلخ). ولكن من الممكن في دفتر ملاحظات.

سوف تحتوي الأمثلة على الفراغات للتنفيذ في الخدمة ، وتنفيذ مكتوب بالفعل. أولاً ، قم بتشغيل التثبيت والتجميع وإلقاء نظرة فاحصة على الخدمات وواجهة برمجة التطبيقات الخاصة بها.

يتم تقديم مثال الخدمات و API نفسها لأغراض التوضيح فقط ؛ لا تقم بنقل كل شيء AS IS إلى المنتج الخاص بك!

أولاً ، استنساخ المستودع بالخدمات المقدمة لأنفسنا ، والتكامل الذي سنفعله به ، والانتقال إلى الدليل:

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

في محطة منفصلة ، نقوم بجمع جميع التطبيقات باستخدام gradle ، حيث سيتم إطلاق جميع الخدمات بعد gradle باستخدام gradle .

 ./gradlew build && docker-compose up 

أثناء تنزيل كل شيء وتثبيته ، فكر في مشروع يحتوي على خدمات.

سيتم استلام طلب يحتوي على رمز وأرقام بطاقات للتحويل والمبلغ المطلوب تحويله بين البطاقات عند مدخل الخدمة (الخدمة التجريبية):

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

وفقًا لرمز authToken ، تحتاج إلى الذهاب إلى خدمة AUTH والحصول على userId ، حيث يمكنك بعد ذلك تقديم طلب إلى USER وسحب جميع المعلومات الإضافية على المستخدم. ستعيد AUTH أيضًا معلومات حول أي من الخدمات الثلاث التي يمكننا الوصول إليها. استجابة عينة من AUTH :

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

للتنقل بين البطاقات ، اذهب أولاً مع كل رقم بطاقة في CARD . استجابة للطلبات ، سوف نتلقى cardId ، ثم نرسل طلبًا إلى PAYMENT ونقوم بإجراء التحويل. والأخير - مرة أخرى ، نرسل طلبًا إلى PAYMENT مع fromCardId ومعرفة الرصيد الحالي.

لمحاكاة تأخير بسيط في الخدمات ، يتم طرح قيمة متغير البيئة TIMEOUT في جميع الحاويات ، حيث يتم تعيين تأخير الاستجابة بالميلي ثانية. ولتنويع الاستجابات من AUTH ، من الممكن تغيير قيمة SUCCESS_RATE ، والتي تتحكم في احتمال الاستجابة true للخدمة.

ملف 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" 

بالنسبة لجميع الخدمات ، تتم إعادة توجيه المنافذ من 8081 إلى 8084 للوصول إليها مباشرة بسهولة.

دعنا ننتقل إلى كتابة Demo service . أولاً ، دعنا نحاول كتابة التطبيق على أنه خرقاء قدر الإمكان ، دون التزامن والتزامن. للقيام بذلك ، خذ Spring boot 2.2.1 و Kotlin وفارغ للخدمة. نقوم باستنساخ المستودع والانتقال إلى فرع spring-mvc-start :

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

انتقل إلى ملف demo.Controller . يحتوي على الأسلوب processRequest الفارغ فقط الذي يجب أن يكتب التنفيذ.

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

سيتم استلام طلب تحويل بين البطاقات عند مدخل الطريقة.

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

لتسهيل الأمر ، تم بالفعل إجراء جميع أساليب التكامل مع الخدمات الأخرى ، تحتاج فقط إلى توصيلها بشكل صحيح. هناك خمس طرق فقط ، واحدة لكل إجراء. يتم تنفيذ المكالمات إلى الخدمات الأخرى نفسها على مكالمة حظر Spring RestTemplate .

طريقة المثال للاتصال 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'") } 

دعنا ننتقل إلى تنفيذ الأسلوب. تشير التعليقات إلى الإجراء والاستجابة المتوقعة في الإخراج:

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

أولاً ، نطبق الطريقة بأقصى قدر ممكن ، دون مراعاة أن AUTH يمكنها أن تمنعنا من الوصول إلى خدمات أخرى. حاول أن تفعل ذلك بنفسك. عندما يتحول (أو بعد التبديل إلى فرع spring-mvc ) ، يمكنك التحقق من تشغيل الخدمة على النحو التالي:

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

بدء الخدمة (من مجلد الخدمة التجريبية):

 ./gradlew bootRun 

نرسل طلبًا لنقطة النهاية:

 ./demo-request.sh 

استجابة لذلك ، حصلنا على شيء مثل هذا:

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

في المجموع ، تحتاج إلى تقديم 6 طلبات من أجل تنفيذ الخدمة. وبالنظر إلى أن كل واحد منهم يستجيب مع تأخير قدره 100 مللي ثانية ، لا يمكن أن يكون إجمالي الوقت أقل من 600 مللي ثانية. في الواقع ، اتضح أن حوالي 700 مللي ثانية ، مع الأخذ في الاعتبار جميع النفقات العامة. حتى الآن الكود بسيط للغاية ، وإذا أردنا الآن إضافة فحص استجابة AUTH للوصول إلى خدمات أخرى ، فلن يكون ذلك صعبًا (مثل أي إعادة بيع أخرى).

ولكن دعونا نفكر في كيفية تسريع تنفيذ الاستعلام. إذا لم تأخذ في الاعتبار التحقق من استجابة AUTH ، فلدينا مهمتان مستقلتان:

• الحصول على userId وطلب البيانات من USER ؛
• تلقي cardId لكل بطاقة ، إجراء الدفع واستلام المبلغ الإجمالي.

يمكن تنفيذ هذه المهام بشكل مستقل عن بعضها البعض. ثم يعتمد وقت التنفيذ الإجمالي على أطول سلسلة من المكالمات (في هذه الحالة ، الثانية) وسيتم تنفيذه في المجموع لمدة 300 مللي + X مللي مقدار الحمل.

بالنظر إلى أن المكالمات نفسها تمنع ، فإن الطريقة الوحيدة لتنفيذ الطلبات المتوازية هي تشغيلها على سلاسل عمليات منفصلة. يمكنك إنشاء سلسلة رسائل منفصلة لكل مكالمة ، ولكنها ستكون مكلفة للغاية. هناك طريقة أخرى لتشغيل المهام على ThreadPool. للوهلة الأولى ، يبدو هذا الحل مناسبًا ، وسيقل الوقت حقًا. على سبيل المثال ، يمكننا تنفيذ الاستعلامات على CompleteableFuture. يتيح لك تشغيل مهام الخلفية عن طريق استدعاء الأساليب مع postfix async . وإذا لم تقم بتحديد ThreadPool معين عند استدعاء الأساليب ، فسيتم تشغيل المهام على ForkJoinPool.commonPool() . حاول كتابة تطبيق بنفسك أو انتقل إلى فرع 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 ) } 

إذا قمنا الآن بقياس وقت الطلب ، فسيكون ذلك في حدود 360 مللي ثانية. بالمقارنة مع الإصدار الأصلي ، انخفض إجمالي الوقت بحوالي 2 مرة. أصبحت الشفرة نفسها أكثر تعقيدًا بعض الشيء ، لكن حتى الآن لا يزال من الصعب تعديلها. وإذا أردنا هنا إضافة فحص استجابة من AUTH ، فهذا ليس بالأمر الصعب.

ولكن ماذا لو كان لدينا عدد كبير من الطلبات الواردة للخدمة نفسها؟ قل حوالي 1000 طلب في وقت واحد؟ مع هذا النهج ، اتضح بسرعة أن جميع مؤشرات الترابط ThreadPool مشغولون بإجراء حظر المكالمات. وقد توصلنا إلى استنتاج مفاده أن الإصدار الحالي أيضًا لا يناسب.

يبقى فقط أن تفعل شيئا مع الخدمة تدعو أنفسهم. يمكنك تعديل الاستعلامات وجعلها غير محظورة. بعد ذلك ، ستُرجع طرق استدعاء الخدمات CompleteableFuture أو Flux أو Observable أو Deferred أو Promise أو كائن مشابه لبناء سلسلة من التوقعات. مع هذا النهج ، لا نحتاج إلى إجراء مكالمات على سلاسل رسائل منفصلة - سيكون يكفي وجود واحد (أو على الأقل مجموعة مؤشرات ترابط منفصلة صغيرة) استعارناها بالفعل لمعالجة الطلبات.

هل يمكننا الآن تحمل العبء الثقيل على الخدمة؟ للإجابة على هذا السؤال ، ألق نظرة فاحصة على Tomcat ، والذي يستخدم في برنامج Spring boot 2.2.1 في بداية org.springframework.boot:spring-boot-starter-web . إنه مصمم بحيث يتم تخصيص مؤشر ترابط من ThreadPool لكل طلب وارد للمعالجة الخاصة به. وفي حالة عدم وجود تدفقات مجانية ، ستصبح الطلبات الجديدة "قائمة انتظار" للانتظار. لكن خدمتنا نفسها ترسل طلبات إلى خدمات أخرى فقط. تخصيص مجرى كامل تحته وحظره حتى تأتي إجابات من الجميع ، ليبدو بعبارة ملطفة.

لحسن الحظ ، سمح Spring مؤخرًا باستخدام خادم ويب غير محظور استنادًا إلى Netty أو Undertow. للقيام بذلك ، ستحتاج فقط إلى تغيير spring-boot-starter-webflux spring-boot-starter-web إلى spring-boot-starter-webflux وتغيير طريقة معالجة الطلبات التي سيتم "تغليف" الطلب والرد عليها بشكل طفيف. هذا يرجع إلى حقيقة أن Webflux مبني على أساس مفاعل ، وبالتالي الآن في الطريقة التي تحتاج إليها لبناء سلسلة من التحولات أحادية.

حاول كتابة تنفيذك غير المحظور للأسلوب. للقيام بذلك ، انتقل إلى فرع spring-webflux-start . يرجى ملاحظة أن مشغل Spring Boot قد تغير ، حيث يتم الآن استخدام الإصدار مع Webflux ، كما تم تغيير تنفيذ الطلبات على الخدمات الأخرى التي تمت إعادة كتابتها لاستخدام WebClient غير WebClient .

طريقة المثال للاتصال AUTH:

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

يتم إدراج تطبيق المثال الأول في محتويات الأسلوب processRequest في التعليق. حاول إعادة كتابتها بنفسك على مفاعل. مثل آخر مرة ، قم أولاً بإنشاء الإصدار دون مراعاة الشيكات من AUTH ، ثم انظر إلى مدى صعوبة إضافتها:

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

بعد التعامل مع هذا ، يمكنك مقارنة تنفيذي من 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 ) } } 

توافق على أن كتابة تنفيذ الآن (مقارنة بنهج الحجب السابق) أصبحت أكثر صعوبة. وإذا أردنا إضافة شيكات "منسية" من AUTH ، فلن يكون ذلك سهلاً.

هذا هو جوهر النهج التفاعلي. انه لشيء رائع لبناء سلاسل معالجة غير مخولة. ولكن في حالة ظهور المتفرعة ، فإن الرمز لم يعد بهذه البساطة.

يمكن أن يساعدك كوريوتين ، Kotlin ، اللطفاء للغاية مع أي كود غير متزامن / تفاعلي ، هنا. بالإضافة إلى ذلك ، هناك عدد كبير من الأغلفة المكتوبة لـ Reactor ، و CompleteableFuture ، إلخ. ولكن حتى إذا لم تجد ما هو مناسب ، فيمكنك دائمًا كتابته بنفسك ، باستخدام شركات بناء خاصة.

دعونا إعادة كتابة التنفيذ على coroutines بمفردنا. للقيام بذلك ، انتقل إلى فرع spring-webflux-coroutines-start . تتم إضافة التبعيات الضرورية إليها في 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") 

وتغيير الأسلوب processRequest :

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

لم تعد بحاجة إلى Mono وترجمتها ببساطة إلى وظيفة تعليق (بفضل تكامل Spring و Kotlin). بالنظر إلى أننا سننشئ coroutines إضافية في الطريقة ، سنحتاج إلى إنشاء coroutineScope الكشفية التابعة (لفهم أسباب إنشاء نطاق إضافي ، راجع وظيفة رومان إليزاروف على التزامن المهيكل ). يرجى ملاحظة أن مكالمات الخدمة الأخرى لم تتغير على الإطلاق. يعيدون نفس Mono ، والتي يمكنك استدعاء طريقة suspend awaitFirst "للانتظار" لنتيجة الاستعلام.

إذا كانت coroutines لا تزال مفهوما جديدا بالنسبة لك ، ثم هناك دليل رائع مع وصف مفصل. حاول كتابة spring-webflux-coroutines الخاص لأسلوب processRequest أو انتقل إلى فرع 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 ) } 

يمكنك مقارنة الكود بالنهج التفاعلي. مع coroutines ، لا يجب عليك التفكير في جميع النقاط الفرعية مقدماً. يمكننا ببساطة استدعاء أساليب await وإيقاف المهام غير async في async في الأماكن الصحيحة. يظل الكود مشابهًا قدر الإمكان للنسخة المباشرة الأصلية ، وهو أمر يصعب تغييره على الإطلاق. والعامل المهم هو أن coroutines هي ببساطة جزءا لا يتجزأ من رمز رد الفعل.

قد تحب الطريقة التفاعلية في هذه المهمة أكثر ، لكن الكثير من الأشخاص الذين شملهم الاستطلاع يجدون صعوبة أكبر. بشكل عام ، كلتا الطريقتين تحلان مشكلتهما ويمكنك استخدام واحدة تريد. بالمناسبة ، في الآونة الأخيرة في Kotlin ، هناك أيضًا فرصة لإنشاء كورينات "باردة" مع Flow ، والتي تشبه إلى حد بعيد Reactor. صحيح ، أنها لا تزال في المرحلة التجريبية ، ولكن الآن يمكنك إلقاء نظرة على التطبيق الحالي وتجربته في الكود.

أريد أن أنهي هنا وأترك ​​روابط مفيدة أخيرًا:

• دليل Coroutine
• مفاعل كوروتين
• Coroutines CompletableFuture
• رومان إليزاروف عن الكوروتين
• تدفقات Kotlin و Coroutines
• التزامن منظم
• التمهيد الربيع يحب Kotlin

أتمنى أن تكون مهتمًا وأنك تمكنت من كتابة طريقة تنفيذية لكل الطرق بنفسك. وبالطبع ، أريد أن أصدق أنك تحب الخيار مع coroutines أكثر =)

شكرا لكل من قرأ حتى النهاية!