获取ES客户端

src/main/scala/tools/es/ESClient.scala

+package tools.es
+
+import org.apache.http.HttpHost
+import org.elasticsearch.client.sniff.Sniffer
+import org.elasticsearch.client.{RestClient, RestClientBuilder, RestHighLevelClient}
+import tools.properties.PropertiesTools
+
+import java.io.IOException
+import java.util.Properties
+import scala.collection.mutable.ArrayBuffer
+
+
+/**
+ * @ClassName: ESClient
+ * @Description: 连接ES客户端
+ * @Create by: LinYang
+ * @Date: 2020/12/17 10:34
+ */
+object ESClient {
+
+  private[this] val properties: Properties = PropertiesTools.getProperties
+
+  //主机和端口
+  private[this] val hostsAndPorts: Array[String] = properties.getProperty("es.cluster").split(",")
+
+  private[this] var highLevelClient: RestHighLevelClient = _
+  private[this] var sniffer: Sniffer = _
+
+  //  构建客户端
+  private[this] def esClient: RestClientBuilder = {
+
+    val httpHosts: ArrayBuffer[HttpHost] = new ArrayBuffer[HttpHost]()
+
+    if (0 != hostsAndPorts.length) {
+      for (elem <- hostsAndPorts) {
+        val hostAndPort: Array[String] = elem.split(":")
+        val host = hostAndPort(0)
+        val port = hostAndPort(1).trim.toInt
+        val httpHost = new HttpHost(host, port)
+        httpHosts.+=(httpHost)
+      }
+    }
+
+    val restClientBuilder = RestClient.builder(httpHosts.toArray: _*)
+      .setFailureListener(new RestClient.FailureListener() {
+        def onFailure(node: Nothing): Unit = {
+          super.onFailure(node)
+        }
+      })
+    //      .setRequestConfigCallback((requestConfigBuilder: RequestConfig.Builder) => requestConfigBuilder.setSocketTimeout(10000))
+    restClientBuilder
+  }
+
+  /**
+   * 获取客户端
+   *
+   * @return 返回ES官方推荐的客户端
+   */
+  def getClient: RestHighLevelClient = {
+    highLevelClient = new RestHighLevelClient(esClient)
+    val lowLevelClient = highLevelClient.getLowLevelClient
+    sniffer = Sniffer.builder(lowLevelClient)
+      .setSniffAfterFailureDelayMillis(10000)
+      .build()
+    highLevelClient
+  }
+
+  /**
+   * 关闭客户端连接
+   */
+  def closeClient(): Unit = {
+    if (null != highLevelClient) {
+      try {
+        sniffer.close()
+        highLevelClient.close()
+      } catch {
+        case e: IOException => e.printStackTrace()
+      }
+    }
+  }
+}