Tingkatkan throughput aplikasi sebanyak 2 kali atau kerja tanpa pemblokiran dengan Elasticsearch menggunakan coroutine Kotlin

Elasticsearch adalah mesin pencari yang kuat dan sistem penyimpanan dokumen terdistribusi. Dengan konfigurasi yang benar, dialah yang melakukan semua keajaiban pencarian, dan aplikasi klien hanya perlu membuat permintaan dalam bentuk DSL Kueri dan menunggu tanggapan.

, , , ? RPS, .

, , !

Elasticsearch. Docker- .

docker run --name elastic -p 9200:9200 --env-file ./docker.env -d docker.elastic.co/elasticsearch/elasticsearch:7.12.0
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

docker.env 2 - single-node 1 .

name:

{
    "_index": "record",
    "_type": "_doc",
    "_id": "_Erb8HkByb6IPnpUYTbS",
    "_score": 1.0,
    "_source": {
        "name": "Bob"
    }
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

Spring Boot . , , :

plugins {
    id("org.springframework.boot") version "2.4.3"
    id ("io.spring.dependency-management") version "1.0.11.RELEASE"
    id("org.jetbrains.kotlin.plugin.spring") version "1.4.20"
    kotlin("jvm") version "1.4.20"
}

dependencies {
    implementation("org.springframework.boot:spring-boot-starter-web")

    //     elasticsearch
    implementation("org.elasticsearch:elasticsearch:$esVersion")
    implementation("org.elasticsearch.client:elasticsearch-rest-client:$esVersion")
    implementation("org.elasticsearch.client:elasticsearch-rest-high-level-client:$esVersion")
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

RestHighLevelClient

fun search(term: String): List<String> {
    val sourceBuilder = SearchSourceBuilder().apply {
         //   ID
         query(QueryBuilders.idsQuery().addIds(term))
    }
    val request = SearchRequest(indexNames, sourceBuilder).apply {
        requestCache(false) //      
    }
    
    return client.search(request, RequestOptions.DEFAULT)
        .let { mapResults(it) }
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

REST query.

:

INFO 22720 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat initialized with port(s): 8080 (http)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

spring-boot-starter-web Tomcat, 1 web-. , .

RPS jmeter - . - Concurrency Thread Group. -, .

:

• Target Concurrency: 250

  
  
  
  
  

• Ramp Up Time: 30

  
  
  
  
  

• Ramp-Up Steps Count: 250

  
  
  
  
  

, 30 , 0 250. .

- Threads VusualVM.

: Address already in use: connect. Windows - jmeter` . .

:

- - , http-nio-8080-exec-* (0-9) 10 199. - , max-threads 200.

, . jmeter - .

, , .

, Summary Report, jmeter`:

, 22 . 7 , - 618. , 717 .

.

, , ? , :

1. query

   
   
   
   
   

2. 
   
   
   
   

3. 
   
   
   
   

4. 
   
   
   
   

5. 
   
   
   
   

: ", , . ? !". , . , elasticsearch-client , !

, . - , . CPS (Continuation-passing style), -, . Roman Elizarov - Deep dive into Kotlin coroutines.

, :

• spring-boot-starter-web



spring-boot-starter-webflux
  
  
  
  

  
  
  
  
  

• . org.jetbrains.kotlinx:kotlinx-coroutines-core, kotlinx-coroutines-reactor kotlinx-coroutines-reactive
  
  
  
  

  
  
  
  
  

• - , suspendCoroutine. , API , , , Continuation, , . :

  
  
  
  
  

suspendCoroutine<SearchResponse> { continuation -> //  
    //  ,        
    val callback = AsyncSearchResponseActionListener(continuation)

    //      
    client.asyncSearch().submitAsync(request, RequestOptions.DEFAULT, callback)
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

class AsyncSearchResponseActionListener(private val continuation: Continuation<SearchResponse>) :
    ActionListener<AsyncSearchResponse> {
    //     Continuatuion
    override fun onResponse(response: AsyncSearchResponse) = continuation.resume(response.searchResponse)

    //     ,    
    override fun onFailure(ex: Exception) = continuation.resumeWithException(ex)
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

• suspendCoroutine, - , suspend.

  
  
  
  
  

:

INFO 9868 --- [           main] o.s.b.web.embedded.netty.NettyWebServer  : Netty started on port 8080
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

web- webflux, , . , , Reactor.

.

. threads:

, , , . , , http- Wait, , Running.

:

:

Kemudian kita dapat menyimpulkan bahwa waktu eksekusi permintaan telah berkurang, sedangkan jumlah tugas yang diselesaikan dan throughput aplikasi, sebaliknya, meningkat. Sekarang, menurut jmeter, jumlah permintaan untuk periode waktu yang sama hampir dua kali lipat - dari 717 menjadi 1259 per detik!

Hasil

Bandwidth aplikasi memainkan peran penting dalam layanan beban tinggi. Untuk memaksimalkan pemanfaatan utas dan, sebagai akibatnya, meningkatkan throughput, Anda dapat bermigrasi ke strategi non-pemblokiran dalam menulis kode. Hal ini dapat dilakukan dengan menggunakan server aplikasi alternatif yang digabungkan dengan implementasi komunikasi asinkron dalam bentuk coroutine Kotlin.