参考文档：
Java High Level REST Client

引用

  1. <dependency>
  2.     <groupId>org.elasticsearch</groupId>
  3.     <artifactId>elasticsearch</artifactId>
  4.     <version>6.3.2</version>
  5. </dependency>
  6. <dependency>
  7.     <groupId>org.elasticsearch.client</groupId>
  8.     <artifactId>transport</artifactId>
  9.     <version>6.3.2</version>
  10. </dependency>
  11. <dependency>
  12.     <groupId>org.elasticsearch.client</groupId>
  13.     <artifactId>elasticsearch-rest-high-level-client</artifactId>
  14.     <version>6.3.2</version>
  15. </dependency>
  16. <!-- https://mvnrepository.com/artifact/commons-beanutils/commons-beanutils -->
  17. <dependency>
  18.     <groupId>commons-beanutils</groupId>
  19.     <artifactId>commons-beanutils</artifactId>
  20.     <version>1.9.3</version>
  21. </dependency>

单个示例

client

  1. RestHighLevelClient client = new RestHighLevelClient(
  2.                 RestClient.builder(
  3.                         new HttpHost("10.20.222.191", 9200, "http")
  4.                 ).build()

x-pack 需要特殊验证

  1.         // 阿里云ES集群需要basic auth验证。
  2.         final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
  3.        //访问用户名和密码为您创建阿里云Elasticsearch实例时设置的用户名和密码，也是Kibana控制台的登录用户名和密码。
  4.         credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("{访问用户名}", "{访问密码}"));
  6.        // 通过builder创建rest client，配置http client的HttpClientConfigCallback。
  7.        // 单击所创建的Elasticsearch实例ID，在基本信息页面获取公网地址，即为ES集群地址。
  8.         RestClientBuilder builder = RestClient.builder(new HttpHost("{ES集群地址}", 9200))
  9.                 .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
  10.                     @Override
  11.                     public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
  12.                         return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
  13.                     }
  14.                 });
  16.         // RestHighLevelClient实例通过REST low-level client builder进行构造。
  17.         RestHighLevelClient highClient = new RestHighLevelClient(builder);

create document

image.png

  1. Map<String, Object> jsonMap = new HashMap<>();
  2. jsonMap.put("city", "北京");
  3. jsonMap.put("state", "beijing");
  4. jsonMap.put("location", new HashMap<String,String>(){{
  5.     put("lat", "39.1");
  6.     put("lon", "116.1");
  7. }});
  8. // 2、设置索引的settings，可以在参数里指定ID
  9. IndexRequest indexRequest = new IndexRequest("city", "doc")
  10.         .source(jsonMap);
  12. IndexResponse indexResponse = client.index(indexRequest);
  13. 》》》》》》》》》》》》》》》》》》》》》》》》》
  14. # 处理结果
  15. String index = indexResponse.getIndex();
  16. String type = indexResponse.getType();
  17. String id = indexResponse.getId();
  18. long version = indexResponse.getVersion();
  19. if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
  20.     # 新建
  21. } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
  22.     # 更新
  23. }
  24. ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
  25. if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
  26.     # 成功的分片个数
  27. }
  28. if (shardInfo.getFailed() > 0) {
  29.     for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
  30.         String reason = failure.reason(); 
  31.   # 失败处理
  32.     }
  33. }

根据id查询

  1. GetRequest getRequest = new GetRequest("city", "doc", "AW7Unt433ydeaETTj2DG");
  2. GetResponse getFields = client.get(getRequest);
  3. Map<String, Object> sourceAsMap = getFields.getSourceAsMap();
  4. 返回：
  5. {city=北京, location={lon=116.1, lat=39.1}, state=beijing}

根据条件查询

  1. SearchRequest request = new SearchRequest("city");
  2. request.types("doc");
  3. SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
  4. QueryBuilders.boolQuery();
  5. MatchAllQueryBuilder allQuery = QueryBuilders.matchAllQuery();
  6. sourceBuilder.query(allQuery);
  8. request.source(sourceBuilder);
  11. SearchResponse searchResponse = client.search(request);
  12. SearchHits hits = searchResponse.getHits();
  13. long totalHits = hits.getTotalHits();
  14. float maxScore = hits.getMaxScore();
  15. SearchHit[] searchHits = hits.getHits();
  16. for (SearchHit hit : searchHits) {
  17.     Map<String, Object> sourceAsMap = hit.getSourceAsMap();
  18.     System.out.println(sourceAsMap);
  19. }
  20.       》》》》》》》》》》》》》》》》
  22. {city=Beijing, location={lon=116.41667, lat=39.91667}, state=BJ}
  23. {city=Shanghai, location={lon=121.43333, lat=34.50000}, state=SH}
  24. {city=Xiamen, location={lon=118.10000, lat=24.46667}, state=FJ}
  25. {city=Fuzhou, location={lon=119.30000, lat=26.08333}, state=FJ}
  26. {city=Guangzhou, location={lon=113.23333, lat=23.16667}, state=GD}
  27. {city=北京, location={lon=116.1, lat=39.1}, state=beijing}

根据ID删除

  1. DeleteRequest deleteRequest = new DeleteRequest("city","doc","AW7Unt433ydeaETTj2DG");
  2. DeleteResponse deleteResponse = client.delete(deleteRequest);
  4. 》》》》》》》》》》》》》》》》》》》》》》》》》》》》》》
  5. String index = deleteResponse.getIndex();
  6. String type = deleteResponse.getType();
  7. String id = deleteResponse.getId();
  8. long version = deleteResponse.getVersion();
  9. ReplicationResponse.ShardInfo shardInfo = deleteResponse.getShardInfo();
  10. if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
  11.     # 处理成功分片数量小于总分片数量的情况
  12. }
  13. if (shardInfo.getFailed() > 0) {
  14.     for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
  15.         String reason = failure.reason(); 
  16.                 # 处理潜在的故障    
  17.     }
  18. }

根据id update

重新赋值

  1. Map<String, Object> jsonMap = new HashMap<>();
  2. jsonMap.put("updated", new Date());
  3. jsonMap.put("reason", "daily update");
  4. UpdateRequest request = new UpdateRequest("posts", "doc", "1")
  5.         .doc(jsonMap);

执行脚本

  1. UpdateRequest updateRequest = new UpdateRequest("city", "doc", "AW7TsQ0W3ydeaETTj2DA");
  2. Map<String, Object> parameters = new HashMap<String, Object>() {{
  3.     put("count", 4); // 可以是数字或者字符串
  4. }};
  5. Script inline = new Script(ScriptType.INLINE, "painless", "ctx._source.count += params.count", parameters);
  6. updateRequest.script(inline);
  8. client.update(updateRequest);
  1. String index = updateResponse.getIndex();
  2. String type = updateResponse.getType();
  3. String id = updateResponse.getId();
  4. long version = updateResponse.getVersion();
  5. if (updateResponse.getResult() == DocWriteResponse.Result.CREATED) {
  7. } else if (updateResponse.getResult() == DocWriteResponse.Result.UPDATED) {
  9. } else if (updateResponse.getResult() == DocWriteResponse.Result.DELETED) {
  10.     # 处理文档被删除的情况
  11. } else if (updateResponse.getResult() == DocWriteResponse.Result.NOOP) {
  12.     # 处理文档没有受到更新影响的情况(没有对文档执行任何操作(noop))
  13. }

批量执行

  1. BulkRequest request = new BulkRequest();
  2. request.add(new DeleteRequest("posts", "doc", "3")); 
  3. request.add(new UpdateRequest("posts", "doc", "2") 
  4.         .doc(XContentType.JSON,"other", "test"));
  5. request.add(new IndexRequest("posts", "doc", "4")  
  6.         .source(XContentType.JSON,"field", "baz"));
  9. BulkResponse bulkResponse = client.bulk(request);
  11. 》》》》》》》》》》》
  13. for (BulkItemResponse bulkItemResponse : bulkResponse) { 
  14.     DocWriteResponse itemResponse = bulkItemResponse.getResponse(); 
  16.     if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.INDEX
  17.             || bulkItemResponse.getOpType() == DocWriteRequest.OpType.CREATE) { 
  18.         IndexResponse indexResponse = (IndexResponse) itemResponse;
  20.     } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) { 
  21.         UpdateResponse updateResponse = (UpdateResponse) itemResponse;
  23.     } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.DELETE) { 
  24.         DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
  25.     }
  26. }

scroll实例

  1. SearchRequest searchRequest = new SearchRequest("posts");
  2. SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
  3. searchSourceBuilder.query(matchQuery("title", "Elasticsearch"));
  4. searchSourceBuilder.size(size); 
  5. searchRequest.source(searchSourceBuilder);
  6. searchRequest.scroll(TimeValue.timeValueMinutes(1L)); 
  7. SearchResponse searchResponse = client.search(searchRequest);
  8. String scrollId = searchResponse.getScrollId(); 
  9. SearchHits hits = searchResponse.getHits();  
  12. 第二次，循环知道结果为空
  14. SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId); 
  15. scrollRequest.scroll(TimeValue.timeValueSeconds(30));
  16. SearchResponse searchScrollResponse = client.searchScroll(scrollRequest);
  17. scrollId = searchScrollResponse.getScrollId();  
  18. hits = searchScrollResponse.getHits(); 
  19. assertEquals(3, hits.getTotalHits());
  20. assertEquals(1, hits.getHits().length);
  21. assertNotNull(scrollId);

low-client发送自定义请求

highclient不支持按条件删除，可以通过low-client来解决

封装实例

FaqRestClient.java

  1. import com.alibaba.fastjson.JSON;
  2. import com.unisound.model.FaqModel;
  3. import org.apache.commons.beanutils.BeanMap;
  4. import org.apache.commons.beanutils.BeanUtils;
  5. import org.apache.http.HttpHost;
  6. import org.apache.http.auth.AuthScope;
  7. import org.apache.http.auth.UsernamePasswordCredentials;
  8. import org.apache.http.client.CredentialsProvider;
  9. import org.apache.http.entity.ContentType;
  10. import org.apache.http.impl.client.BasicCredentialsProvider;
  11. import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
  12. import org.apache.http.nio.entity.NStringEntity;
  13. import org.apache.http.util.EntityUtils;
  14. import org.elasticsearch.action.bulk.BulkRequest;
  15. import org.elasticsearch.action.bulk.BulkResponse;
  16. import org.elasticsearch.action.delete.DeleteRequest;
  17. import org.elasticsearch.action.delete.DeleteResponse;
  18. import org.elasticsearch.action.index.IndexRequest;
  19. import org.elasticsearch.action.index.IndexResponse;
  20. import org.elasticsearch.action.search.SearchRequest;
  21. import org.elasticsearch.action.search.SearchResponse;
  22. import org.elasticsearch.action.support.replication.ReplicationResponse;
  23. import org.elasticsearch.action.update.UpdateRequest;
  24. import org.elasticsearch.action.update.UpdateResponse;
  25. import org.elasticsearch.client.Response;
  26. import org.elasticsearch.client.RestClient;
  27. import org.elasticsearch.client.RestClientBuilder;
  28. import org.elasticsearch.client.RestHighLevelClient;
  29. import org.elasticsearch.search.SearchHit;
  30. import org.elasticsearch.search.SearchHits;
  31. import org.elasticsearch.search.builder.SearchSourceBuilder;
  32. import org.springframework.stereotype.Repository;
  34. import javax.annotation.PostConstruct;
  35. import java.io.IOException;
  36. import java.util.ArrayList;
  37. import java.util.HashMap;
  38. import java.util.List;
  39. import java.util.Map;
  41. @Repository
  42. public class FaqRestClient {
  43.     private String ip;
  44.     private Integer port;
  45.     private String userName;
  46.     private String password;
  48.     private String index;
  49.     private String type;
  51.     private RestHighLevelClient client;
  54.     @PostConstruct
  55.     private void init() {
  56.         // 阿里云ES集群需要basic auth验证。
  57.         final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
  58.         //访问用户名和密码为您创建阿里云Elasticsearch实例时设置的用户名和密码，也是Kibana控制台的登录用户名和密码。
  59.         credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(userName, password));
  61.         // 通过builder创建rest client，配置http client的HttpClientConfigCallback。
  62.         // 单击所创建的Elasticsearch实例ID，在基本信息页面获取公网地址，即为ES集群地址。
  63.         RestClientBuilder builder = RestClient.builder(new HttpHost(this.ip, this.port))
  64.                 .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
  65.                     @Override
  66.                     public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
  67.                         return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
  68.                     }
  69.                 });
  71.         // RestHighLevelClient实例通过REST low-level client builder进行构造。
  72.         client = new RestHighLevelClient(builder);
  73.     }
  75.     /**
  76.      * 返回结果是否为真
  77.      *
  78.      * @param shardInfo
  79.      * @return
  80.      */
  81.     private Boolean getSuccess(ReplicationResponse.ShardInfo shardInfo) {
  82.         if (shardInfo.getFailed() > 0) {
  83.             for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
  84.                 String reason = failure.reason();
  85.             }
  86.             return false;
  87.         }
  88.         return true;
  89.     }
  91.     /**
  92.      * 插入一个model
  93.      *
  94.      * @param faqModel
  95.      * @return
  96.      * @throws IOException
  97.      */
  98.     public Boolean insert(FaqModel faqModel) throws IOException {
  99.         IndexRequest indexRequest = new IndexRequest(index, type)
  100.                 .source(new BeanMap(faqModel));
  101.         IndexResponse indexResponse = client.index(indexRequest);
  103.         return getSuccess(indexResponse.getShardInfo());
  105.     }
  107.     /**
  108.      * 批量插入
  109.      *
  110.      * @param faqModels
  111.      * @return
  112.      * @throws IOException
  113.      */
  114.     public Boolean inserts(List<FaqModel> faqModels) throws IOException {
  115.         BulkRequest bulkRequest = new BulkRequest();
  116.         for (FaqModel faqModel : faqModels) {
  117.             bulkRequest.add(new IndexRequest(index, type).source(new BeanMap(faqModel)));
  118.         }
  119.         BulkResponse bulkResponse = client.bulk(bulkRequest);
  120.         return !bulkResponse.hasFailures();
  121.     }
  123.     /**
  124.      * 根据id删除
  125.      *
  126.      * @param id
  127.      * @return
  128.      * @throws IOException
  129.      */
  130.     public Boolean deleteById(String id) throws IOException {
  131.         DeleteRequest deleteRequest = new DeleteRequest(index, type, id);
  132.         DeleteResponse deleteResponse = client.delete(deleteRequest);
  134.         return getSuccess(deleteResponse.getShardInfo());
  135.     }
  137.     /**
  138.      * 批量删除
  139.      *
  140.      * @param ids
  141.      * @return
  142.      * @throws IOException
  143.      */
  144.     public Boolean deletes(List<String> ids) throws IOException {
  145.         BulkRequest bulkRequest = new BulkRequest();
  146.         for (String id : ids) {
  147.             bulkRequest.add(new DeleteRequest(index, type, id));
  148.         }
  149.         BulkResponse bulkResponse = client.bulk(bulkRequest);
  150.         return !bulkResponse.hasFailures();
  151.     }
  153.     /**
  154.      * 查询
  155.      *
  156.      * @param sourceBuilder
  157.      * @return
  158.      * @throws IOException
  159.      */
  160.     public List<FaqModel> query(SearchSourceBuilder sourceBuilder) throws Exception {
  161.         SearchRequest searchRequest = new SearchRequest(index);
  162.         searchRequest.types(type);
  163.         searchRequest.source(sourceBuilder);
  164.         SearchResponse searchResponse = client.search(searchRequest);
  165.         SearchHits hits = searchResponse.getHits();
  166.         SearchHit[] searchHits = hits.getHits();
  167.         ArrayList<FaqModel> faqModels = new ArrayList<>();
  168.         for (SearchHit hit : searchHits) {
  169.             Map<String, Object> sourceAsMap = hit.getSourceAsMap();
  170.             FaqModel faqModel = new FaqModel();
  171.             BeanUtils.populate(faqModel, sourceAsMap);
  172.             faqModels.add(faqModel);
  173.         }
  174.         return faqModels;
  175.     }
  177.     /**
  178.      * 根据id更新
  179.      *
  180.      * @param faqModel
  181.      * @return
  182.      * @throws IOException
  183.      */
  184.     public Boolean updateById(FaqModel faqModel) throws IOException {
  185.         UpdateRequest updateRequest = new UpdateRequest(index, type, faqModel.getId().toString())
  186.                 .doc(new BeanMap(faqModel));
  187.         UpdateResponse updateResponse = client.update(updateRequest);
  188.         return getSuccess(updateResponse.getShardInfo());
  189.     }
  191.     /**
  192.      * 根据字段精确查询删除
  193.      * {
  194.      * "query":{
  195.      * "terms":{
  196.      * "key":""
  197.      * }
  198.      * }
  199.      * }
  200.      *
  201.      * @param key
  202.      * @param list
  203.      * @return
  204.      */
  205.     public Integer deleteByQuery(String key, List<String> list) throws Exception {
  206.         String jsonStr = "{\n" +
  207.                 "    \"query\":{\n" +
  208.                 "        \"terms\":{\n" +
  209.                 "            \"" + key + "\": " + JSON.toJSONString(list) + "\n" +
  210.                 "        }\n" +
  211.                 "    }\n" +
  212.                 "}";
  213.         RestClient lowLevelClient = client.getLowLevelClient();
  214.         HashMap<String, String> param = new HashMap<String, String>() {{
  215.             put("conflicts", "proceed");
  216.         }};
  217.         NStringEntity entity = new NStringEntity(jsonStr, ContentType.APPLICATION_JSON);
  218.         Response performResponse = lowLevelClient.performRequest("POST", "/" + index + "/" + type + "/delete_by_query", param, entity);
  219.         return JSON.parseObject(EntityUtils.toString(performResponse.getEntity())).getInteger("deleted");
  220.     }
  221. }