ES工具类

1、POM依赖

    <dependencies>
        <dependency>
            <groupId>io.searchbox</groupId>
            <artifactId>jest</artifactId>
            <version>5.3.3</version>

        </dependency>

        <dependency>
            <groupId>net.java.dev.jna</groupId>
            <artifactId>jna</artifactId>
            <version>4.5.2</version>
        </dependency>

        <dependency>
            <groupId>org.codehaus.janino</groupId>
            <artifactId>commons-compiler</artifactId>
            <version>2.7.8</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.elasticsearch/elasticsearch -->
        <dependency>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch</artifactId>
            <version>2.4.6</version>
        </dependency>
    </dependencies>

2、工具类

import java.util.Objects

import io.searchbox.client.config.HttpClientConfig
import io.searchbox.client.{JestClient, JestClientFactory}
import io.searchbox.core.{Bulk, Index}

/**
 * Author z
 * Date 2020-08-27 09:19:55
 */
object ESUtil {
  private var factory: JestClientFactory = _
  private val ES_HOST = "http://hadoop300"
  private val ES_HTTP_PORT = 9200
  
  /**
   * 获取客户端
   *
   * @return 客户端对象
   */
  def getClient = {
    if (!Objects.isNull(factory))
      build()
    factory.getObject
  }
  
  /**
   * 关闭客户端对象
   *
   * @param c
   */
  def close(c: JestClient): Unit = {
    if (!Objects.isNull(c))
      try
        c.close()
      catch {
        case e: Exception => e.printStackTrace()
      }
  }
  
  /**
   * 构建连接
   */
  private def build() = {
    factory = new JestClientFactory
    factory.setHttpClientConfig(
      new HttpClientConfig.Builder(ES_HOST + ":" + ES_HTTP_PORT)
        .multiThreaded(true) //允许多线程
        .maxTotalConnection(20) //最大连接数
        .connTimeout(10000) //连接超时时间
        .readTimeout(10000) //读取超时时间
        .build()
    )
  }
  
  def insertBulk(indexName: String, list: List[Any]) = {
    val jest = getClient
    val builder = new Bulk.Builder()
      .defaultIndex(indexName).defaultType("_doc")
    for (doc <- list) {
      val index = new Index.Builder(doc).build()
      builder.addAction(index)
    }
    val items = jest.execute(builder.build()).getItems
    println(s"保存=${items.size()}")
    close(jest)
  }
  def main(args: Array[String]): Unit = {
    val jest: JestClient = getClient
    val  source="{
  "name":"li4",
  "age":456,
  "amount": 250.1,
  "phone_num":"138***0003"
}"
    val index: Index = new Index.Builder(source).index("test").`type`("_doc").build()
    jest.execute(index)
    close(jest)
  }
}