Increase application throughput by 2 times or non-blocking work with Elasticsearch using Kotlin coroutines

Elasticsearch is a powerful search engine and distributed document storage system. With the correct configuration, it is he who performs all the search magic, and the client application only needs to generate a request in the form of a Query DSL and wait for a response.

, , , ? 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.

:

:

Then we can conclude that the execution time of requests has decreased, while the total number of completed tasks and the application throughput, on the contrary, have increased. Now, according to jmeter, the number of requests has almost doubled over the same period of time - from 717 to 1259 per second!

Outcome

Application bandwidth plays an important role in high-load services. In order to maximize utilization of threads and, as a consequence, increase throughput, you can migrate to a non-blocking strategy in writing code. This can be done using alternative application servers coupled with asynchronous communication implementations in the form of Kotlin coroutines.