【ES】Java High Level REST Client 使用示例（增加修改）

ES提供了多種編程語言的鏈接方式，有Java API，PHP API，.NET API

官網可以詳細了解

https://www.elastic.co/guide/en/elasticsearch/client/index.html 

下面闡述ES支持的客戶端鏈接方式：

1：REST API

可以通過瀏覽器請求get方法進行鏈接；

利用Postman等工具發起REST請求；

Java發起HttpClient請求；

2：Transport鏈接

通過socket鏈接,用官網一個的TransPort客戶端，底層是netty

特別提示：

ES在7.0版本開始將廢棄TransportClient，8.0版本開始將完全移除TransportClient

取而代之的是High Level REST Client。

Java High Level REST Client 為高級別的Rest客戶端，基於低級別的REST客戶端，增加了編組請求JSON串，解析響應JSON串等相關API，使用的版本需要和ES服務端的版本保持一致，否則會有版本問題。

首先在使用Java REST Client的時候引入maven的Jar包依賴：

<dependency>
   <groupId>org.elasticsearch.client</groupId>
   <artifactId>elasticsearch-rest-high-level-client</artifactId>
   <version>6.2.4</version>
</dependency>

　　

將log4j2.xml編譯到classes路徑下

<?xml version="1.0" encoding="UTF-8"?>
<configuration status="OFF">
   <appenders>
       <Console name="Console" target="SYSTEM_OUT">
           <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
       </Console>
   </appenders>
   <loggers>
       <root level="info">
           <appender-ref ref="Console"/>
       </root>
   </loggers>
</configuration>

　　

Java High Level REST Client 初始化以及增加修改，批量操作使用示例：

package com.herbert.test;

import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.rest.RestStatus;

import java.io.IOException;
import java.util.*;

/**
* Created by Herbert on 2019/1/28.
*/
public class ESUtil {

   protected RestHighLevelClient client;

   /**
    * Java High Level REST Client  初始化
    */
   public ESUtil (){
       client = new RestHighLevelClient(
               RestClient.builder(
                       new HttpHost("localhost", 9200, "http")));
   }

   /**
    *  新增，修改文檔
    * @param indexName  索引
    * @param type mapping type
    * @param id 文檔id
    * @param jsonStr 文檔數據
    */
   public void addData(String indexName,String type ,String id,String jsonStr) {
       try {
           // 1、創建索引請求  //索引  // mapping type  //文檔id
           IndexRequest request = new IndexRequest(indexName, type, id);     //文檔id
           // 2、准備文檔數據
           // 直接給JSON串
           request.source(jsonStr, XContentType.JSON);
           //4、發送請求
           IndexResponse indexResponse = null;
           try {
               // 同步方式
               indexResponse = client.index(request);
           } catch (ElasticsearchException e) {
               // 捕獲，並處理異常
               //判斷是否版本沖突、create但文檔已存在沖突
               if (e.status() == RestStatus.CONFLICT) {
                   System.out.println("沖突了，請在此寫沖突處理邏輯！" + e.getDetailedMessage());
               }
           }
           //5、處理響應
           if (indexResponse != null) {
               String index1 = indexResponse.getIndex();
               String type1 = indexResponse.getType();
               String id1 = indexResponse.getId();
               long version1 = indexResponse.getVersion();
               if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
                   System.out.println("新增文檔成功!" + index1 + type1 + id1 + version1);
               } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
                   System.out.println("修改文檔成功!");
               }
               // 分片處理信息
               ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
               if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                   System.out.println("分片處理信息.....");
               }
               // 如果有分片副本失敗，可以獲得失敗原因信息
               if (shardInfo.getFailed() > 0) {
                   for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                       String reason = failure.reason();
                       System.out.println("副本失敗原因：" + reason);
                   }
               }
           }

       } catch (Exception e) {
           e.printStackTrace();
       }
   }

   /**
    *  批量插入ES
    * @param indexName 索引
    * @param type 類型
    * @param idName id名稱
    * @param list 數據集合
    */
   public void bulkDate(String indexName,String type ,String idName ,List<Map<String,Object>> list ){
       try {

           if(null == list || list.size()<=0){
               return;
           }
           if(StringUtils.isBlank(indexName) || StringUtils.isBlank(idName)  || StringUtils.isBlank(type)){
               return;
           }
           BulkRequest request = new BulkRequest();
           for(Map<String,Object> map : list){
               if(map.get(idName)!=null){
                   request.add(new IndexRequest(indexName, type, String.valueOf(map.get(idName)))
                           .source(map,XContentType.JSON));
               }
           }
           // 2、可選的設置
           /*
           request.timeout("2m");
           request.setRefreshPolicy("wait_for");
           request.waitForActiveShards(2);
           */
           //3、發送請求
           // 同步請求
           BulkResponse bulkResponse = client.bulk(request);
           //4、處理響應
           if(bulkResponse != null) {
               for (BulkItemResponse bulkItemResponse : bulkResponse) {
                   DocWriteResponse itemResponse = bulkItemResponse.getResponse();

                   if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.INDEX
                           || bulkItemResponse.getOpType() == DocWriteRequest.OpType.CREATE) {
                       IndexResponse indexResponse = (IndexResponse) itemResponse;
                       //TODO 新增成功的處理
                       System.out.println("新增成功,{}"+ indexResponse.toString());
                   } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) {
                       UpdateResponse updateResponse = (UpdateResponse) itemResponse;
                       //TODO 修改成功的處理
                       System.out.println("修改成功,{}"+ updateResponse.toString());
                   } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.DELETE) {
                       DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
                       //TODO 刪除成功的處理
                       System.out.println("刪除成功,{}"+ deleteResponse.toString());
                   }
               }
           }
       } catch (IOException e) {
           e.printStackTrace();
       }
   }
   public static void main(String ags[]){
       Map<String,Object> map1 = new HashMap<String, Object>();
       map1.put("id","2");
       map1.put("user1","bbherbert1");
       map1.put("postDate","2018-08-30");
       map1.put("username","aa");
       map1.put("message","message");
       Map<String,Object> map2 = new HashMap<String, Object>();
       map2.put("id","3");
       map2.put("user2","bbherbert1");
       map2.put("postDate","2018-08-30");
       map2.put("username","aa");
       map2.put("message","message");
       Map<String,Object> map = new HashMap<String, Object>();
       map.put("id","1");
       map.put("user","bbherbert1");
       map.put("postDate","2018-08-30");
       map.put("username","aa");
       map.put("message","message");

       List<Map<String,Object>> list = new ArrayList<Map<String, Object>>();
       list.add(map);
       list.add(map1);
       list.add(map2);
       ESUtil esUtil= new ESUtil();
       esUtil.bulkDate("book15","boo","id",list);
//        Map<String,Object> map = new HashMap<String, Object>();
//        map.put("user","herbert1");
//        map.put("postDate","2018-08-30");
//        map.put("username","aa");
//        map.put("message","message");
//        String jsonString = JSON.toJSONString(map);
//        esUtil.addData("hh","d","4",jsonString);
//        esUtil.addData("hh","d","4","{" +
//                "\"user\":\"kimchy\"," +
//                "\"postDate\":\"2013-01-30\"," +
//                "\"username\":\"zhangsan\"," +
//                "\"message\":\"trying out Elasticsearch\"" +
//                "}");
   }

}

 

上述簡單的介紹了增加文檔的操作示例。詳情可參照官方API