Augmentez le débit des applications par 2 ou travaillez non bloquant avec Elasticsearch à l'aide des coroutines Kotlin

Elasticsearch est un puissant moteur de recherche et un système de stockage de documents distribué. Avec la bonne configuration, c'est lui qui effectue toute la magie de la recherche, et l'application cliente n'a plus qu'à générer une requête sous la forme d'un Query DSL et attendre une réponse.

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

:

:

On peut alors conclure que le temps d'exécution des requêtes a diminué, tandis que le nombre total de tâches terminées et le débit de l'application, au contraire, ont augmenté. Maintenant, selon jmeter, le nombre de requêtes pour la même période a presque doublé - de 717 à 1259 par seconde !

Résultat

La bande passante des applications joue un rôle important dans les services à forte charge. Afin de maximiser l'utilisation des threads et, par conséquent, d'augmenter le débit, vous pouvez migrer vers une stratégie non bloquante en écriture de code. Cela peut être fait en utilisant des serveurs d'applications alternatifs couplés à des implémentations de communication asynchrones sous la forme de coroutines Kotlin.