👮 ⬜️ 👨🏻‍🏭 वर्टेक्स में अतुल्यकालिक प्रोग्रामिंग के तीन प्रतिमान 💛 👵🏾 🍯

मैं एसिंक्रोनस प्रोग्रामिंग के तीन प्रतिमान दिखाना चाहता हूं - कॉलबैक, वायदा, कोरटाइन, वर्टेक्स ढांचे पर एक साधारण वेब एप्लिकेशन के उदाहरण का उपयोग करके। हम कोड को कोटलिन में लिखेंगे।

मान लीजिए कि हमारे पास एक एप्लिकेशन है जो HTTP अनुरोध में एक निश्चित स्ट्रिंग प्राप्त करता है, इसका उपयोग करने वाले डेटाबेस में एक URL की खोज करता है, उस URL पर जाता है और अपनी सामग्री को क्लाइंट को वापस भेजता है।
वर्टेक्स को अत्यधिक लोड किए गए अनुप्रयोगों के लिए एक अतुल्यकालिक ढांचे के रूप में कल्पना की जाती है, जो कि नेट्टी, नए आईओ, इवेंट बस का उपयोग करता है

जैसा कि वर्टेक्स में प्रथागत है, एक वर्टिकल (अभिनेता का एक एनालॉग, यदि आप अक्का जानते हैं) एक अनुरोध प्राप्त करता है, कुछ अन्य BusinessVerticle के URL की खोज के लिए एक इवेंट बस स्ट्रिंग भेजता है, जो वास्तविक काम में लगा हुआ है।

object Main { @JvmStatic fun main(args: Array<String>) { val vertx = Vertx.vertx() vertx.deployVerticle(HttpVerticle()) vertx.deployVerticle(BusinessVerticle()) } }

 class HttpVerticle : AbstractVerticle() { @Throws(Exception::class) override fun start(startFuture: Future<Void>) { val router = createRouter() vertx.createHttpServer() .requestHandler(router) .listen(8080) { result -> if (result.succeeded()) { startFuture.complete() } else { startFuture.fail(result.cause()) } } } private fun createRouter(): Router = Router.router(vertx).apply { get("/").handler(handlerRoot) } private val handlerRoot = Handler<RoutingContext> { rc -> vertx.eventBus().send("my.addr", rc.request().getParam("id") ?: "") { resp: AsyncResult<Message<String>> -> if (resp.succeeded()) { rc.response().end(resp.result().body()) } else { rc.fail(500) } } } }

मानक एपीआई में, सभी अतुल्यकालिक कॉलबैक के माध्यम से किया जाता है, इसलिए BusinessVerticle का प्रारंभिक कार्यान्वयन इस तरह दिखेगा:

 class BusinessVerticle : AbstractVerticle() { private lateinit var dbclient: JDBCClient private lateinit var webclient: WebClient override fun start() { vertx.eventBus().consumer<String>("my.addr") { message -> handleMessage(message) } dbclient = JDBCClient.createShared( vertx, JsonObject() .put("url", "jdbc:postgresql://localhost:5432/payroll") .put("driver_class", "org.postgresql.Driver") .put("user", "vala") .put("password", "vala") .put("max_pool_size", 30) ) val options = WebClientOptions() .setUserAgent("My-App/1.2.3") options.isKeepAlive = false webclient = WebClient.create(vertx, options) } private fun handleMessage(message: Message<String>) { dbclient.getConnection { res -> if (res.succeeded()) { val connection = res.result() connection.query("SELECT url FROM payee_company where name='${message.body()}'") { res2 -> if (res2.succeeded()) { try { val url = res2.result().rows[0].getString("url").removePrefix("http://") webclient .get(url,"/") .send { ar -> if (ar.succeeded()) { val response = ar.result() message.reply(response.bodyAsString()) } else { message.fail(500, ar.cause().message) } } } catch (e: Exception) { message.fail(500, e.message) } } else { message.fail(500, res2.cause().message) } } } else { message.fail(500, res.cause().message) } } } }

यह स्पष्ट रूप से, ऐसा लगता है - कॉलबैक नरक, विशेष रूप से त्रुटि से निपटने।

आइए स्थिति को सुधारने का प्रयास करें, क्योंकि कॉलबैक गुरु हमें सिखाता है - प्रत्येक कॉलबैक को एक अलग विधि से चुनकर:

  class BusinessVerticle: AbstractVerticle() { private lateinit var dbclient: JDBCClient private lateinit var webclient: WebClient override fun start() { vertx.eventBus().consumer<String>("my.addr") { message -> handleMessage(message) } dbclient = JDBCClient.createShared( vertx, JsonObject() .put("url", "jdbc:postgresql://localhost:5432/payroll") .put("driver_class", "org.postgresql.Driver") .put("user", "vala") .put("password", "vala") .put("max_pool_size", 30) ) val options = WebClientOptions() .setUserAgent("My-App/1.2.3") options.isKeepAlive = false webclient = WebClient.create(vertx, options) } private fun handleMessage(message: Message<String>) { dbclient.getConnection { res -> handleConnectionCallback(res, message) } } private fun handleConnectionCallback( res: AsyncResult<SQLConnection>, message: Message<String> ) { if (res.succeeded()) { val connection = res.result() connection.query("SELECT url FROM payee_company where name='${message.body()}'") { res2 -> handleQueryCallBack(res2, message) } } else { message.fail(500, res.cause().message) } } private fun handleQueryCallBack( res2: AsyncResult<ResultSet>, message: Message<String> ) { if (res2.succeeded()) { try { val url = res2.result().rows[0].getString("url").removePrefix("http://") webclient .get(url, "/") .send { ar -> handleHttpCallback(ar, message) } } catch (e: Exception) { message.fail(500, e.message) } } else { message.fail(500, res2.cause().message) } } private fun handleHttpCallback( ar: AsyncResult<HttpResponse<Buffer>>, message: Message<String> ) { if (ar.succeeded()) { // Obtain response val response = ar.result() message.reply(response.bodyAsString()) } else { message.fail(500, ar.cause().message) } } }

खैर, यह बेहतर हो गया। लेकिन ऐसा भी है।

बहुत सी लाइनें, विशेष रूप से पढ़ने योग्य कोड नहीं, आपको प्रतिक्रिया के लिए संदेश ऑब्जेक्ट को खींचने की आवश्यकता है, कोड पर फैली हुई त्रुटि हैंडलिंग।

आइए फ्यूचर्स का उपयोग करके इस बकवास को फिर से लिखने की कोशिश करें
फ्यूचर्स विशेष रूप से अच्छे हैं क्योंकि फ्यूचर.कॉम () का उपयोग करके उन्हें आसानी से जोड़ा जा सकता है

सबसे पहले, हम मानक वर्टेक्स विधियों का अनुवाद करेंगे जो कॉलबैक स्वीकार करते हैं और भविष्य में लौटने वाले तरीकों में कुछ भी नहीं लौटाते हैं।

हम मौजूदा कक्षाओं में विधियों को जोड़ने के लिए कोटलिन की क्षमता का लाभ उठाते हैं:

 fun JDBCClient.getConnectionF(): Future<SQLConnection> { val f = Future.future<SQLConnection>() getConnection { res -> if (res.succeeded()) { val connection = res.result() f.complete(connection) } else { f.fail(res.cause()) } } return f } fun SQLConnection.queryF(query:String): Future<ResultSet> { val f = Future.future<ResultSet>() query(query) { res -> if (res.succeeded()) { val resultSet = res.result() f.complete(resultSet) } else { f.fail(res.cause()) } } return f } fun <T,M> HttpRequest<T>.sendF(): Future<HttpResponse<M>> { val f = Future.future<HttpResponse<M>>() send() { res -> if (res.succeeded()) { val response = res.result() f.complete(response) } else { f.fail(res.cause()) } } return f }

और इसमें हमारे BusinessVerticle.handleMessage को चालू करें:

  private fun handleMessage(message: Message<String>) { val content = getContent(message) content.setHandler{res-> if (res.succeeded()) { // Obtain response val response = res.result() message.reply(response) } else { message.fail(500, res.cause().message) } } } private fun getContent(message: Message<String>): Future<String> { val connection = dbclient.getConnectionF() val resultSet = connection.compose { it.queryF("SELECT url FROM payee_company where name='${message.body()}'") } val url = resultSet.map { it.rows[0].getString("url").removePrefix("http://") } val httpResponse = url.compose { webclient.get(it, "/").sendF() } val content = httpResponse.map { it.bodyAsString() } return content }

यह अच्छा लग रहा है।

सरल, पठनीय कोड। एक स्थान पर हैंडलिंग में त्रुटि। यदि आवश्यक हो, तो आप अलग-अलग अपवादों पर एक अलग प्रतिक्रिया कर सकते हैं, या, कह सकते हैं, इसे एक अलग फ़ंक्शन में डाल सकते हैं।

कवि का सपना!

लेकिन क्या होता है अगर हमें फ्यूचर्स चेन को किसी शर्त से समाप्त करने की आवश्यकता होती है?
उदाहरण के लिए, यदि डेटाबेस में संबंधित प्रविष्टि नहीं है, तो हम एक अपवाद (और कोड 500 ग्राहक को) नहीं फेंकना चाहते हैं, लेकिन कोड 200 के साथ स्ट्रिंग "नो रिकॉर्ड" लौटाएं।

Future.compose () से श्रृंखला को समाप्त करने का एकमात्र तरीका (जो मुझे पता है) एक अपवाद फेंकने के लिए है।

यानी आपको ऐसा कुछ करने की आवश्यकता है: अपने अपवाद का प्रकार निर्धारित करें, इस अपवाद को फेंक दें यदि डेटाबेस में कोई प्रविष्टि नहीं है, तो इस अपवाद को विशेष तरीके से संभालें।

 class NoContentException(message:String):Exception(message) private fun getContent(message: Message<String>): Future<String> { val connection = dbclient.getConnectionF() val resultSet = connection.compose { it.queryF("SELECT url FROM payee_company where name='${message.body()}'") } val url = resultSet.map { if (it.numRows<1) throw NoContentException("No records") it.rows[0].getString("url").removePrefix("http://") } val httpResponse = url.compose { webclient.get(it, "/").sendF() } val content = httpResponse.map { it.bodyAsString() } return content } private fun handleMessage(message: Message<String>) { val content = getContent(message) content.setHandler{res-> if (res.succeeded()) { // Obtain response val response = res.result() message.reply(response) } else { if (res.cause() is NoContentException) message.reply(res.cause().message) else message.fail(500, res.cause().message) } } }

यह काम करता है!

लेकिन यह पहले से ही बदतर लग रहा है - निष्पादन के प्रवाह को नियंत्रित करने के लिए अपवादों का उपयोग करना सुंदर नहीं है। और अगर कई ऐसे मामले होंगे जिनके लिए अलग प्रसंस्करण की आवश्यकता होती है, तो कोड बहुत कम पठनीय बन जाएगा।

आइए कोटलिन कोरटाइन के साथ भी ऐसा ही करने की कोशिश करते हैं।
कॉर्बिन के बारे में बहुत कुछ लिखा गया है, जिसमें हैबे ( 1 , 2 , 3 , ...) भी शामिल है, इसलिए मैंने उनके बारे में अलग से नहीं लिखा।

वर्टेक्स के नवीनतम संस्करण स्वचालित रूप से सभी विधियों के कॉरआउट संस्करण उत्पन्न करते हैं जिन्हें कॉलबैक को स्वीकार करना चाहिए।

हम पुस्तकालयों को जोड़ते हैं
'Vertx-lang-kotlin-coroutines'
'Vertx-lang-kotlin'

और, उदाहरण के लिए

 JDBCClient.getConnectionAwait() SQLConnection.queryAwait()

आदि

तब हमारे संदेश प्रसंस्करण के तरीके कुछ अच्छे और सरल में बदल जाते हैं:

 private suspend fun handleMessage(message: Message<String>) { try { val content = getContent(message) message.reply(content) } catch(e:Exception){ message.fail(500, e.message) } } private suspend fun getContent(message: Message<String>): String { val connection = dbclient.getConnectionAwait() val resultSet = connection.queryAwait("SELECT url FROM payee_company where name='${message.body()}'") val url = resultSet.rows[0].getString("url").removePrefix("http://") val httpResponse = webclient.get(url, "/").sendAwait() val content = httpResponse.bodyAsString() return content }

ठीक है, आपको कॉलआउट संदर्भ प्रदान करके कॉल को बदलना होगा:

 vertx.eventBus().consumer<String>("my.addr") { message -> GlobalScope.launch(vertx.dispatcher()) { handleMessage(message)} }

यहाँ क्या हो रहा है?

अद्वैत के साथ ये सभी विधियां कोड को अतुल्यकालिक रूप से बुलाती हैं, इसके परिणाम की प्रतीक्षा करें, और जब वे प्रतीक्षा करते हैं, तो थ्रेड किसी अन्य कोरआउट के निष्पादन पर स्विच हो जाता है।

यदि हम हुड के नीचे देखते हैं, तो यह इस तरह दिखता है:

 suspend fun SQLClient.getConnectionAwait(): SQLConnection { return awaitResult { this.getConnection(it) } } suspend fun <T> awaitResult(block: (h: Handler<AsyncResult<T>>) -> Unit): T { val asyncResult = awaitEvent(block) if (asyncResult.succeeded()) return asyncResult.result() else throw asyncResult.cause() } suspend fun <T> awaitEvent(block: (h: Handler<T>) -> Unit): T { return suspendCancellableCoroutine { cont: CancellableContinuation<T> -> try { block.invoke(Handler { t -> cont.resume(t) }) } catch (e: Exception) { cont.resumeWithException(e) } } }

फ्यूचर्स के साथ हमारे स्व-लिखित कार्यान्वयन के समान।

लेकिन यहाँ हम सामान्य कोड प्राप्त करते हैं - रिटर्न टाइप (भविष्य के बजाय) के रूप में स्ट्रिंग, AsyncResult के साथ बदसूरत कॉलबैक के बजाय / पकड़ें

और अगर हमें निष्पादन श्रृंखला को बीच में रोकने की आवश्यकता है, तो यह बिना किसी अपवाद के स्वाभाविक दिखता है:

  private suspend fun getContent(message: Message<String>): String { val connection = dbclient.getConnectionAwait() val resultSet = connection.queryAwait("SELECT url FROM payee_company where name='${message.body()}'") if (resultSet.numRows<1) return "No records" val url = resultSet.rows[0].getString("url").removePrefix("http://") val httpResponse = webclient.get(url, "/").sendAwait() val content = httpResponse.bodyAsString() return content }

मेरी राय में, महान!

वर्टेक्स में अतुल्यकालिक प्रोग्रामिंग के तीन प्रतिमान

More articles: