Java之Elasticsearch 增刪改查

<!--ELK  -->
<dependency>
   <groupId>org.elasticsearch.client</groupId>
   <artifactId>transport</artifactId>
   <version>5.1.1</version>
   <exclusions>
      <!--<exclusion>
         <artifactId>transport-netty4-client</artifactId>
         <groupId>org.elasticsearch.plugin</groupId>
      </exclusion>-->
   </exclusions>
</dependency>
<dependency>
   <groupId>org.apache.logging.log4j</groupId>
   <artifactId>log4j-core</artifactId>
   <version>2.6.2</version>
</dependency>

<!-- springmvc json -->
<dependency>
   <groupId>com.fasterxml.jackson.core</groupId>
   <artifactId>jackson-databind</artifactId>
   <version>2.8.8</version>
</dependency>
<dependency>
   <groupId>org.codehaus.jackson</groupId>
   <artifactId>jackson-mapper-asl</artifactId>
   <version>1.9.13</version>
</dependency>
<dependency>
   <groupId>com.fasterxml.jackson.core</groupId>
   <artifactId>jackson-core</artifactId>
   <version>2.8.8</version>
</dependency>
<dependency>
   <groupId>com.fasterxml.jackson.core</groupId>
   <artifactId>jackson-annotations</artifactId>
   <version>2.8.5</version>
</dependency>

package com.sxis.util;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;

import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

/**
  * 【描述】：  ELK java API 接口 ，包括查詢，刪除，插入等待
  * 【步驟】：     
  * @param * @return * @throws * @author Allen
  * @date 2017/7/4 9:47
  */
public class ElasticsearchUtil {
   private static TransportClient client;
   private static String elasticIp = "192.168.211.50";
   private static int elasticPort = 9300;


   /**
     * 【描述】：  初始化ElasticSearch對象
     * 【步驟】：
     * @param * @return * @throws * @author Allen
     * @date 2017/7/4 15:19
     */
public static void init() throws UnknownHostException, InterruptedException, ExecutionException {
      //設置ES實例的名稱.put("client.transport.sniff", true) //自動嗅探整個集群的狀態，把集群中其他ES節點的ip添加到本地的客戶端列表中
      Settings esSettings = Settings.builder().put("cluster.name", "elasticsearch").build();
      client = new PreBuiltTransportClient(esSettings);//初始化client較老版本發生了變化，此方法有幾個重載方法，初始化插件等。
      //此步驟添加IP，至少一個，其實一個就夠了，因為添加了自動嗅探配置
      client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(elasticIp), elasticPort));
      System.out.println("連接建立成功");
   }


   /**
     * 【描述】：  創建index，把其中的文檔轉化為json的格式存儲
     * 【步驟】：
     * @param 　　<Node>:節點ip <port>:節點端口號，默認9200 <Index>:索引名 <Type>:索引類型 <ID>:操作對象的ID號
    * @return * @throws * @author Allen
     * @date 2017/7/5 9:42
     */
public static void createIndex() throws ElasticsearchException,IOException {
      for (int i=300; i<=50000000;i++){
         User user = new User();
         user.setId(1);
         user.setName("huang fox " + i);
         user.setAge(i % 100);
         IndexResponse indexResponse = null;
         indexResponse = client.prepareIndex("users", "user",i+"").setSource(generateJson(user)).execute().actionGet();
         System.out.println("responseIsCreated: "+indexResponse);
      }
      System.out.println("it is ok ！");
   }

   public static void query() throws Exception {
      //term查詢
      //QueryBuilder queryBuilder = QueryBuilders.termQuery("age", 50) ;  //年齡等於50
      //range查詢
      QueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("age").gt(50); //年齡大於50
      SearchResponse searchResponse = client.prepareSearch("users")
            .setTypes("user")
            .setQuery(rangeQueryBuilder)     //query
            .setPostFilter(QueryBuilders.rangeQuery( "age" ).from( 70 ).to( 80 )) // Filter
            .addSort("age", SortOrder.DESC)
            .setSize(120)   // 不設置的話，默認取10條數據
            .execute().actionGet();
      SearchHits hits = searchResponse.getHits();
      System.out.println("查到記錄數："+hits.getTotalHits());
      SearchHit[] searchHists = hits.getHits();
      if(searchHists.length>0){
         for(SearchHit hit:searchHists){
            String name =  (String) hit.getSource().get("name");
            Integer age = Integer.parseInt( hit.getSource().get("age").toString() );
            System.out.format("name:%s ,age :%d \n",name ,age);
         }
      }
   }


   /**
    * 轉換成json對象
    *
    * @param user
* @return */
private static String generateJson(User user) {
      String json = "";
      try {
         XContentBuilder contentBuilder = XContentFactory.jsonBuilder().startObject();
         contentBuilder.field("id", user.getId());
         contentBuilder.field("name", user.getName());
         contentBuilder.field("age", user.getAge());
         json = contentBuilder.endObject().string();
      } catch (IOException e) {
         e.printStackTrace();
      }
      return json;
   }


   /*
    * Get index 獲取文檔相當於讀取數據庫的一行數據
    */
   public static void getIndex(){
      GetResponse getresponse = client.prepareGet("users", "user", "402").execute().actionGet();
      System.out.println(getresponse.getSourceAsString());
   }

   /*
    *Delete index 相當於刪除一行數據
    */
   public static void delete(){
      DeleteResponse deleteresponse = client.prepareDelete("users", "user","1").execute().actionGet();
      System.out.println(deleteresponse.getVersion());
   }

   /*
    *Delete index 刪除索引及該索引所管轄的記錄
    */
   public static void deleteIndex(){
      //刪除所有記錄
      DeleteIndexResponse deleteIndexResponse = client.admin().indices().prepareDelete("logs").execute().actionGet();
      System.out.println(deleteIndexResponse.isAcknowledged());
   }

   /**
     * 【描述】：     獲取到所有的索引
     * 【步驟】：
     * @param * @return * @throws * @author Allen
     * @date 2017/7/4 16:27
     */
public static void getAllIndex(){
      ClusterStateResponse response = client.admin().cluster().prepareState().execute().actionGet();
      //獲取所有索引
      String[] indexs=response.getState().getMetaData().getConcreteAllIndices();
      for (String index : indexs) {
         System.out.println( index + " delete" );//
      }
   }



   public void close(){
      //on shutdown 斷開集群
      client.close();
   }

   public static void main( String[] args ) {
      try {
         init();
         //createIndex();
         //getIndex();
         //delete();
         //deleteIndex();
         //getAllIndex();
         query();
      } catch( Exception e ) {
         e.printStackTrace();
      }
   }

}


class User{

   private static final long serialVersionUID = 5290251719938640641L;

   private Integer id;
   private String name;
   private int age;

   public Integer getId() {
      return id;
   }

   public void setId( Integer id ) {
      this.id = id;
   }

   public String getName() {
      return name;
   }

   public void setName( String name ) {
      this.name = name;
   }

   public int getAge() {
      return age;
   }

   public void setAge( int age ) {
      this.age = age;
   }
}