使用Java High Level REST Client操作elasticsearch

说明

在明确了ES的基本概念和使用方法后，我们来学习如何使用ES的Java API.
本文假设你已经对ES的基本概念已经有了一个比较全面的认识。

客户端

你可以用Java客户端做很多事情：

• 执行标准的index,get,delete,update,search等操作。
• 在正在运行的集群上执行管理任务。

但是，通过官方文档可以得知，现在存在至少三种Java客户端。

1. Transport Client
2. Java High Level REST Client
3. Java Low Level Rest Client

造成这种混乱的原因是：

• 长久以来，ES并没有官方的Java客户端，并且Java自身是可以简单支持ES的API的，于是就先做成了TransportClient。但是TransportClient的缺点是显而易见的，它没有使用RESTful风格的接口，而是二进制的方式传输数据。

• 之后ES官方推出了Java Low Level REST Client,它支持RESTful，用起来也不错。但是缺点也很明显，因为TransportClient的使用者把代码迁移到Low Level REST Client的工作量比较大。官方文档专门为迁移代码出了一堆文档来提供参考。

• 现在ES官方推出Java High Level REST Client,它是基于Java Low Level REST Client的封装，并且API接收参数和返回值和TransportClient是一样的，使得代码迁移变得容易并且支持了RESTful的风格，兼容了这两种客户端的优点。当然缺点是存在的，就是版本的问题。ES的小版本更新非常频繁，在最理想的情况下，客户端的版本要和ES的版本一致（至少主版本号一致），次版本号不一致的话，基本操作也许可以，但是新API就不支持了。

• 强烈建议ES5及其以后的版本使用Java High Level REST Client。笔者这里使用的是ES5.6.3，下面的文章将基于JDK1.8+Spring Boot+ES5.6.3 Java High Level REST Client+Maven进行示例。

前置条件：

• JDK1.8
• elasticsearch 6.3.2（其他版本未做测试，不保证完全兼容）
• maven 
• spring boot

• 1.maven依赖：

        <!--elasticsearch base-->
        <dependency>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch</artifactId>
            <version>6.3.2</version>
        </dependency>
        <!-- Java Low Level REST Client -->
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-client</artifactId>
            <version>6.3.2</version>
        </dependency>
        <!-- Java High Level REST Client -->
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>6.3.2</version>
        </dependency>

• 2.接入rest-higl-level-client

import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ElasticsearchConfiguration implements FactoryBean<RestHighLevelClient>, InitializingBean, DisposableBean {
    private static final Logger LOGGER = LoggerFactory.getLogger(ElasticsearchConfiguration.class);

    @Value("${spring.data.elasticsearch.host}")
    private String host;
    @Value("${spring.data.elasticsearch.port}")
    private int port;
    @Value("${spring.data.elasticsearch.username}")
    private String username;
    @Value("${spring.data.elasticsearch.password}")
    private String password;

    private RestHighLevelClient restHighLevelClient;

    @Override
    public void destroy() throws Exception {
        try {
            LOGGER.info("Closing elasticSearch client");
            if (restHighLevelClient != null) {
                restHighLevelClient.close();
            }
        } catch (final Exception e) {
            LOGGER.error("Error closing ElasticSearch client: ", e);
        }
    }

    @Override
    public RestHighLevelClient getObject() throws Exception {
        return restHighLevelClient;
    }

    @Override
    public Class<RestHighLevelClient> getObjectType() {
        return RestHighLevelClient.class;
    }

    @Override
    public boolean isSingleton() {
        return false;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        buildClient();
    }

    protected void buildClient() {
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));
        RestClientBuilder builder = RestClient.builder(new HttpHost(host, port))
                .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                    @Override
                    public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                        return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                    }
                });

        restHighLevelClient = new RestHighLevelClient(builder);
    }

}

• 3.index api

Map<String, Object> jsonMap = new HashMap<>();
jsonMap.put("user", "laimailai");
jsonMap.put("postDate", new Date());
jsonMap.put("message", "trying out Elasticsearch");
IndexRequest indexRequest = new IndexRequest("index", "type", "1")
        .source(jsonMap);
IndexResponse indexResponse = client.index(request);

• 4.get api

GetRequest getRequest = new GetRequest(
        "index",
        "type",
        "1");
GetResponse getResponse = client.get(request);

• 5.update api

UpdateRequest request = new UpdateRequest(
        "index",
        "type",
        "1");
UpdateResponse updateResponse = client.update(request);

• 6.delete api

DeleteRequest request = new DeleteRequest(
        "index",    
        "type",     
        "1");

• 7.bulk api

之前的文档说明过，bulk接口是批量index/update/delete操作
在API中，只需要一个bulk request就可以完成一批请求。

//1.bulk
BulkRequest request = new BulkRequest();
request.add(new IndexRequest("index", "type", "1")
        .source(XContentType.JSON, "field", "foo"));
request.add(new IndexRequest("index", "type", "2")
        .source(XContentType.JSON, "field", "bar"));
request.add(new IndexRequest("index", "type", "3")
        .source(XContentType.JSON, "field", "baz"));

//同步
BulkResponse bulkResponse = client.bulk(request);

//异步
client.bulkAsync(request, new ActionListener<BulkResponse>() {
    @Override
    public void onResponse(BulkResponse bulkResponse) {

    }

    @Override
    public void onFailure(Exception e) {

    }
});

• 8.bulkprocessor 划重点！！！

BulkProcessor 简化bulk API的使用，并且使整个批量操作透明化。
BulkProcessor 的执行需要三部分组成：

1. RestHighLevelClient :执行bulk请求并拿到响应对象。
2. BulkProcessor.Listener：在执行bulk request之前、之后和当bulk response发生错误时调用。
3. ThreadPool：bulk request在这个线程池中执行操作，这使得每个请求不会被挡住，在其他请求正在执行时，也可以接收新的请求。

示例代码：

@Service
public class ElasticSearchUtil {
    private static final Logger LOGGER = LoggerFactory.getLogger(ElasticSearchUtil.class);

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    private BulkProcessor bulkProcessor;

    @PostConstruct
    public void init() {
        BulkProcessor.Listener listener = new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
                //重写beforeBulk,在每次bulk request发出前执行,在这个方法里面可以知道在本次批量操作中有多少操作数
                int numberOfActions = request.numberOfActions();
                LOGGER.info("Executing bulk [{}] with {} requests", executionId, numberOfActions);
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                //重写afterBulk方法，每次批量请求结束后执行，可以在这里知道是否有错误发生。
                if (response.hasFailures()) {
                    LOGGER.error("Bulk [{}] executed with failures,response = {}", executionId, response.buildFailureMessage());
                } else {
                    LOGGER.info("Bulk [{}] completed in {} milliseconds", executionId, response.getTook().getMillis());
                }
                BulkItemResponse[] responses = response.getItems();
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                //重写方法，如果发生错误就会调用。
                LOGGER.error("Failed to execute bulk", failure);
            }
        };

        //在这里调用build()方法构造bulkProcessor,在底层实际上是用了bulk的异步操作
        BulkProcessor bulkProcessor = BulkProcessor.builder(restHighLevelClient::bulkAsync, listener)
                // 1000条数据请求执行一次bulk
                .setBulkActions(1000)
                // 5mb的数据刷新一次bulk
                .setBulkSize(new ByteSizeValue(5L, ByteSizeUnit.MB))
                // 并发请求数量, 0不并发, 1并发允许执行
                .setConcurrentRequests(0)
                // 固定1s必须刷新一次
                .setFlushInterval(TimeValue.timeValueSeconds(1L))
                // 重试5次，间隔1s
                .setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1L), 5))
                .build();
        this.bulkProcessor = bulkProcessor;
    }

    @PreDestroy
    public void destroy() {
        try {
            bulkProcessor.awaitClose(30, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            LOGGER.error("Failed to close bulkProcessor", e);
        }
        LOGGER.info("bulkProcessor closed!");
    }

    /**
     * 修改
     *
     * @param request
     * @throws IOException
     */
    public void update(UpdateRequest request) {
        this.bulkProcessor.add(request);
    }

    /**
     * 新增
     *
     * @param request
     */
    public void insert(IndexRequest request) {
        this.bulkProcessor.add(request);
    }
}

bulkProcessor使用用例:

        //新建三个 index 请求
        IndexRequest one = new IndexRequest("posts", "doc", "1").
                source(XContentType.JSON, "title", "In which order are my Elasticsearch queries executed?");
        IndexRequest two = new IndexRequest("posts", "doc", "2")
                .source(XContentType.JSON, "title", "Current status and upcoming changes in Elasticsearch");
        IndexRequest three = new IndexRequest("posts", "doc", "3")
                .source(XContentType.JSON, "title", "The Future of Federated Search in Elasticsearch");
        //新的三条index请求加入到上面配置好的bulkProcessor里面。
        bulkProcessor.add(one);
        bulkProcessor.add(two);
        bulkProcessor.add(three);
        // add many request here.
        //bulkProcess必须被关闭才能使上面添加的操作生效
        bulkProcessor.close(); //立即关闭
        //关闭bulkProcess的两种方法：
        try {
            //2.调用awaitClose.
            //简单来说，就是在规定的时间内，是否所有批量操作完成。全部完成，返回true,未完成返//回false
            
            boolean terminated = bulkProcessor.awaitClose(30L, TimeUnit.SECONDS);
            
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

• 9.upsert api

update --当id不存在时将会抛出异常:

UpdateRequest request = new UpdateRequest(index, type, "1").doc(jsonMap);
UpdateResponse response = restHighLevelClient.update(request);

upsert--id不存在时就插入:

UpdateRequest request = new UpdateRequest(index, type, "1").doc(jsonMap).upsert(jsonMap);
UpdateResponse response = restHighLevelClient.update(request);

• 10.search api

Search API提供了对文档的查询和聚合的查询。
它的基本形式:

SearchRequest searchRequest = new SearchRequest();  //构造search request .在这里无参，查询全部索引
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();//大多数查询参数要写在searchSourceBuilder里 
searchSourceBuilder.query(QueryBuilders.matchAllQuery());//增加match_all的条件。

SearchRequest searchRequest = new SearchRequest("posts"); //指定posts索引
searchRequest.types("doc"); //指定doc类型

使用SearchSourceBuilder

大多数的查询控制都可以使用SearchSourceBuilder实现。
举一个简单例子:

SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); //构造一个默认配置的对象
sourceBuilder.query(QueryBuilders.termQuery("user", "kimchy")); //设置查询
sourceBuilder.from(0); //设置从哪里开始
sourceBuilder.size(5); //每页5条
sourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS)); //设置超时时间

配置好searchSourceBuilder后，将它传入searchRequest里：

SearchRequest searchRequest = new SearchRequest();
searchRequest.source(sourceBuilder);

//全量搜索
SearchRequest searchRequest = new SearchRequest();
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
searchSourceBuilder.query(QueryBuilders.matchAllQuery());
searchRequest.source(searchSourceBuilder);
SearchRequest searchRequest = new SearchRequest("index");

//根据多个条件搜索
BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
for (String id: ids) {
    TermQueryBuilder termQueryBuilder = new TermQueryBuilder("id", id);
    boolQueryBuilder.should(termQueryBuilder);
}
SearchRequest searchRequest = new SearchRequest(index);
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
searchSourceBuilder.query(boolQueryBuilder);
searchRequest.source(searchSourceBuilder);
SearchResponse response = null;
    response = restHighLevelClient.search(searchRequest);
return response;

• 11.search scroll api

//scroll 分页搜索
final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1L));
SearchRequest searchRequest = new SearchRequest("posts");
searchRequest.scroll(scroll);
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
searchSourceBuilder.query(matchQuery("title", "Elasticsearch"));
searchRequest.source(searchSourceBuilder);

SearchResponse searchResponse = client.search(searchRequest);
String scrollId = searchResponse.getScrollId();
SearchHit[] searchHits = searchResponse.getHits().getHits();

while (searchHits != null && searchHits.length > 0) {
    SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
    scrollRequest.scroll(scroll);
    searchResponse = client.searchScroll(scrollRequest);
    scrollId = searchResponse.getScrollId();
    searchHits = searchResponse.getHits().getHits();

}

ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
clearScrollRequest.addScrollId(scrollId);
ClearScrollResponse clearScrollResponse = client.clearScroll(clearScrollRequest);
boolean succeeded = clearScrollResponse.isSucceeded();

• 12.排序

SearchSourceBuilder可以添加一种或多种SortBuilder。
有四种特殊的排序实现：

• field
• score
• GeoDistance
• scriptSortBuilder

sourceBuilder.sort(new ScoreSortBuilder().order(SortOrder.DESC)); //按照score倒序排列
sourceBuilder.sort(new FieldSortBuilder("_uid").order(SortOrder.ASC));  //并且按照id正序排列

• 13.过滤

默认情况下，searchRequest返回文档内容，与REST API一样，这里你可以重写search行为。例如，你可以完全关闭"_source"检索。

sourceBuilder.fetchSource(false);

该方法还接受一个或多个通配符模式的数组，以更细粒度地控制包含或排除哪些字段。

String[] includeFields = new String[] {"title", "user", "innerObject.*"};
String[] excludeFields = new String[] {"_type"};
sourceBuilder.fetchSource(includeFields, excludeFields);

• 14.聚合

通过配置适当的 AggregationBuilder ，再将它传入SearchSourceBuilder里,就可以完成聚合请求了。
之前的文档里面，我们通过下面这条命令，导入了一千条account信息:

curl -H "Content-Type: application/json" -XPOST 'localhost:9200/bank/account/_bulk?pretty&refresh' --data-binary "@accounts.json"

随后，我们介绍了如何通过聚合请求进行分组:

GET /bank/_search?pretty
{
  "size": 0,
  "aggs": {
    "group_by_state": {
      "terms": {
        "field": "state.keyword"
      }
    }
  }
}

我们将这一千条数据根据state字段分组，得到响应：

{
  "took": 2,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 999,
    "max_score": 0,
    "hits": []
  },
  "aggregations": {
    "group_by_state": {
      "doc_count_error_upper_bound": 20,
      "sum_other_doc_count": 770,
      "buckets": [
        {
          "key": "ID",
          "doc_count": 27
        },
        {
          "key": "TX",
          "doc_count": 27
        },
        {
          "key": "AL",
          "doc_count": 25
        },
        {
          "key": "MD",
          "doc_count": 25
        },
        {
          "key": "TN",
          "doc_count": 23
        },
        {
          "key": "MA",
          "doc_count": 21
        },
        {
          "key": "NC",
          "doc_count": 21
        },
        {
          "key": "ND",
          "doc_count": 21
        },
        {
          "key": "MO",
          "doc_count": 20
        },
        {
          "key": "AK",
          "doc_count": 19
        }
      ]
    }
  }
}

Java实现:

   @Test
    public void test2(){
        RestClient lowLevelRestClient = RestClient.builder(
                new HttpHost("172.16.73.50", 9200, "http")).build();
        RestHighLevelClient client =
                new RestHighLevelClient(lowLevelRestClient);
        SearchRequest searchRequest = new SearchRequest("bank");
        searchRequest.types("account");
        TermsAggregationBuilder aggregation = AggregationBuilders.terms("group_by_state")
                .field("state.keyword");
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.aggregation(aggregation);
        searchSourceBuilder.size(0);
        searchRequest.source(searchSourceBuilder);
        try {
            SearchResponse searchResponse = client.search(searchRequest);
            System.out.println(searchResponse.toString());
        } catch (IOException e) {
            e.printStackTrace();
        }
        
    }

Search response

Search response返回对象与其在API里的一样，返回一些元数据和文档数据。
首先，返回对象里的数据十分重要，因为这是查询的返回结果、使用分片情况、文档数据,HTTP状态码等

RestStatus status = searchResponse.status();
TimeValue took = searchResponse.getTook();
Boolean terminatedEarly = searchResponse.isTerminatedEarly();
boolean timedOut = searchResponse.isTimedOut();

其次，返回对象里面包含关于分片的信息和分片失败的处理:

int totalShards = searchResponse.getTotalShards();
int successfulShards = searchResponse.getSuccessfulShards();
int failedShards = searchResponse.getFailedShards();
for (ShardSearchFailure failure : searchResponse.getShardFailures()) {
    // failures should be handled here
}

取回searchHit

为了取回文档数据，我们要从search response的返回对象里先得到searchHit对象：

SearchHits hits = searchResponse.getHits();

取回文档数据：

    @Test
    public void test2(){
        RestClient lowLevelRestClient = RestClient.builder(
                new HttpHost("172.16.73.50", 9200, "http")).build();
        RestHighLevelClient client =
                new RestHighLevelClient(lowLevelRestClient);
        SearchRequest searchRequest = new SearchRequest("bank");
        searchRequest.types("account");
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchRequest.source(searchSourceBuilder);
        try {
            SearchResponse searchResponse = client.search(searchRequest);
            SearchHits searchHits = searchResponse.getHits();
            SearchHit[] searchHit = searchHits.getHits();
            for (SearchHit hit : searchHit) {
                System.out.println(hit.getSourceAsString());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        
    }

根据需要，还可以转换成其他数据类型:

String sourceAsString = hit.getSourceAsString();
Map<String, Object> sourceAsMap = hit.getSourceAsMap();
String documentTitle = (String) sourceAsMap.get("title");
List<Object> users = (List<Object>) sourceAsMap.get("user");
Map<String, Object> innerObject = (Map<String, Object>) sourceAsMap.get("innerObject");

取回聚合数据

聚合数据可以通过SearchResponse返回对象，取到它的根节点，然后再根据名称取到聚合数据。

GET /bank/_search?pretty
{
  "size": 0,
  "aggs": {
    "group_by_state": {
      "terms": {
        "field": "state.keyword"
      }
    }
  }
}

响应:

{
  "took": 2,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 999,
    "max_score": 0,
    "hits": []
  },
  "aggregations": {
    "group_by_state": {
      "doc_count_error_upper_bound": 20,
      "sum_other_doc_count": 770,
      "buckets": [
        {
          "key": "ID",
          "doc_count": 27
        },
        {
          "key": "TX",
          "doc_count": 27
        },
        {
          "key": "AL",
          "doc_count": 25
        },
        {
          "key": "MD",
          "doc_count": 25
        },
        {
          "key": "TN",
          "doc_count": 23
        },
        {
          "key": "MA",
          "doc_count": 21
        },
        {
          "key": "NC",
          "doc_count": 21
        },
        {
          "key": "ND",
          "doc_count": 21
        },
        {
          "key": "MO",
          "doc_count": 20
        },
        {
          "key": "AK",
          "doc_count": 19
        }
      ]
    }
  }
}

Java实现:

    @Test
    public void test2(){
        RestClient lowLevelRestClient = RestClient.builder(
                new HttpHost("172.16.73.50", 9200, "http")).build();
        RestHighLevelClient client =
                new RestHighLevelClient(lowLevelRestClient);
        SearchRequest searchRequest = new SearchRequest("bank");
        searchRequest.types("account");
        TermsAggregationBuilder aggregation = AggregationBuilders.terms("group_by_state")
                .field("state.keyword");
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.aggregation(aggregation);
        searchSourceBuilder.size(0);
        searchRequest.source(searchSourceBuilder);
        try {
            SearchResponse searchResponse = client.search(searchRequest);
            Aggregations aggs = searchResponse.getAggregations();
            Terms byStateAggs = aggs.get("group_by_state");
            Terms.Bucket b = byStateAggs.getBucketByKey("ID"); //只取key是ID的bucket
            System.out.println(b.getKeyAsString()+","+b.getDocCount());
            System.out.println("!!!");
            List<? extends Bucket> aggList = byStateAggs.getBuckets();//获取bucket数组里所有数据
            for (Bucket bucket : aggList) {
                System.out.println("key:"+bucket.getKeyAsString()+",docCount:"+bucket.getDocCount());;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

参考：https://www.jianshu.com/p/5cb91ed22956

参考：https://my.oschina.net/u/3795437/blog/2253366