Java High Level REST Client 使用教程

說明

　　之前寫過一個用jest操作es的文章，不過感覺寫的有點亂。而且jest操作es的時候其實很多東西還是使用的es的一些東西。感覺還不如直接用官方的java客戶端了。

所以就研究了下high-level-client，感覺挺好用的，而且官方文檔很詳細。推薦使用。

　　下面進入正題，寫的東西大部分都是基於官方文檔的。自己封裝了個查詢用的Criteria類，方便查詢。

elasticsearch官方文檔: https://www.elastic.co/guide/en/elasticsearch/reference/current/index.html 

Java High Level REST Client 官方文檔 https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/java-rest-high.html 下面的實現大部分基於這個文檔。

es客戶端谷歌插件，方便沒法FQ的同學。 ElasticSearch-Head_v0.1.4.crx https://lanzous.com/iccd9sj

maven引用

        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>7.6.2</version>
        </dependency>
      <!-- 以下非必須引用，是一些輔助工具包 -->
        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.3.2</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.12</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>compile</scope>
        </dependency>

操作

1、主要有 模版、index的一些操作。

2、數據操作方面有 插入、批量插入、刪除、更新、根據條件更新。

3、數據查詢 按數據庫查詢來說有 = (TermQuery) 、<= >= (RangeQuery)、in (TermsQuery)、not in (mustNot+TermsQuery) 都封裝到了Criteria類里。

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.lang.Console;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.json.JSONConfig;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.Data;
import org.apache.http.HttpHost;
import org.elasticsearch.action.admin.indices.alias.Alias;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.*;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.*;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.sort.SortOrder;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.Collections;
import java.util.Date;
import java.util.List;

/**
 * @author: sun
 * @create: 2020-04-27
 **/
public class Main {
    RestHighLevelClient client = null;

    @Before
    public void before() {
        client = new RestHighLevelClient(
                RestClient.builder(
                        new HttpHost("127.0.0.1", 9200, "http")));
    }

    @Test
    public void putTemplate() throws IOException {
        PutIndexTemplateRequest request = new PutIndexTemplateRequest("test_template");
        //別名，所有根據該模版創建的index 都會添加這個別名。查詢時可查詢別名，就可以同時查詢多個名稱不同的index，根據此方法可實現index每天或每月生成等邏輯。
        request.alias(new Alias("test_index"));
        request.order(10);
        //匹配哪些index。在創建index時會生效。
        request.patterns(CollUtil.newArrayList("test_index*"));
        request.settings(Settings.builder()
                //數據插入后多久能查到，實時性要求高可以調低
                .put("index.refresh_interval", "10s")
                //傳輸日志，對數據安全性要求高的 設置 request，默認值:request
                .put("index.translog.durability", "async")
                .put("index.translog.sync_interval", "120s")
                //分片數量
                .put("index.number_of_shards", "5")
                //副本數量
                .put("index.number_of_replicas", "0")
                //單次最大查詢數據的數量。默認10000。不要設置太高，如果有導出需求可以根據查詢條件分批次查詢。
                .put("index.max_result_window", "100000"));
        //使用官方提供的工具構建json。可以直接拼接一個json字符串，也可以使用map嵌套。
        XContentBuilder jsonMapping = XContentFactory.jsonBuilder();
        //所有數據類型 看官方文檔:https://www.elastic.co/guide/en/elasticsearch/reference/7.4/mapping-types.html#_core_datatypes
        jsonMapping.startObject().startObject("properties")
                .startObject("testId").field("type", "long").endObject()
                .startObject("price").field("type", "double").endObject()
                //keyword類型不會分詞存儲
                .startObject("name").field("type", "keyword").endObject()
                //指定分詞器
                .startObject("content").field("type", "text").field("analyzer", "ik_max_word").endObject()
                .startObject("createTime").field("type", "date").field("format", "yyyy-MM-dd HH:mm:ss").endObject()
                .endObject().endObject();
        request.mapping(jsonMapping);
        //設置為true只強制創建，而不是更新索引模板。如果它已經存在，它將失敗
        request.create(false);
        AcknowledgedResponse response = client.indices().putTemplate(request, RequestOptions.DEFAULT);
        if (response.isAcknowledged()) {
            Console.log("創建模版成功!");
        } else {
            Console.log("創建模版失敗!");
        }
    }

    @Test
    public void getTemplate() throws IOException {
        GetIndexTemplatesRequest getIndexTemplatesRequest = new GetIndexTemplatesRequest("test*");
        GetIndexTemplatesResponse getIndexTemplatesResponse = client.indices().getIndexTemplate(getIndexTemplatesRequest, RequestOptions.DEFAULT);
        List<IndexTemplateMetaData> indexTemplates = getIndexTemplatesResponse.getIndexTemplates();
        indexTemplates.forEach(t -> {
            Console.log(t.name());
        });
    }

    @Test
    public void createIndex() throws IOException {
        CreateIndexRequest request = new CreateIndexRequest("test_index_tmp");
        //這里也可以針對index單獨設置。不過因為我們已經設置過模版了，所以這里就不進行設置了。
        //index其實也不用單獨創建，在插入數據時，如果所有不存在，會自動創建索引。
        //request.settings();
        //request.mapping();
        //request.alias()
        CreateIndexResponse createIndexResponse = client.indices().create(request, RequestOptions.DEFAULT);
        if (createIndexResponse.isAcknowledged()) {
            Console.log("創建index成功!");
        } else {
            Console.log("創建index失敗!");
        }
    }

    @Test
    public void getIndex() throws IOException {
        GetIndexRequest request = new GetIndexRequest("test_index*");
        GetIndexResponse response = client.indices().get(request, RequestOptions.DEFAULT);
        String[] indices = response.getIndices();
        for (String indexName : indices) {
            Console.log("index name:{}", indexName);
        }
    }

    @Test
    public void delIndex() throws IOException {
        DeleteIndexRequest request = new DeleteIndexRequest("test_index*");
//        DeleteIndexRequest request = new DeleteIndexRequest("test_index_tmp");
        AcknowledgedResponse response = client.indices().delete(request, RequestOptions.DEFAULT);
        if (response.isAcknowledged()) {
            Console.log("刪除index成功!");
        } else {
            Console.log("刪除index失敗!");
        }
    }

    @Test
    public void insertData() throws IOException {
        //插入數據，index不存在則自動根據匹配到的template創建。index沒必要每天創建一個，如果是為了靈活管理，最低建議每月一個 yyyyMM。
        IndexRequest request = new IndexRequest("test_index_" + DateUtil.format(new Date(), "yyyyMM"));
        //最好不要自定義id 會影響插入速度。
//        request.id("id");
        TestData testData = new TestData();
        testData.setTestId(RandomUtil.randomLong(9999999999L));
        testData.setPrice(10.0D);
        testData.setName(RandomUtil.randomString(8));
        testData.setContent("");
        testData.setCreateTime(new Date());
        request.source(new JSONObject(testData, new JSONConfig().setDateFormat(DatePattern.NORM_DATETIME_PATTERN)).toString()
                , XContentType.JSON);
        IndexResponse response = client.index(request, RequestOptions.DEFAULT);
        Console.log(response);
    }
    @Test
    public void getById() throws IOException {
        //注意 這里查詢使用的是別名。
        GetRequest request = new GetRequest("test_index", "C-54v3EB_Nn045D7VGjz");
        String[] includes = {"name","price"};
        String[] excludes = Strings.EMPTY_ARRAY;
        FetchSourceContext fetchSourceContext = new FetchSourceContext(true,includes,excludes);
        //只查詢特定字段。如果需要查詢所有字段則不設置該項。
        request.fetchSourceContext(fetchSourceContext);
        GetResponse response = client.get(request, RequestOptions.DEFAULT);
        Console.log(response);
        Console.log(JSONUtil.toBean(response.getSourceAsString(),TestData.class));
    }
    
    @Test
    public void delById() throws IOException {
        DeleteRequest request = new DeleteRequest("test_index", "7e5gv3EB_Nn045D7pGdA");
        DeleteResponse response = client.delete(request, RequestOptions.DEFAULT);
        Console.log(response);
    }
    @Test
    public void multiGetById() throws IOException {
        //多個根據id查詢
        MultiGetRequest request = new MultiGetRequest();
        request.add("test_index","D-56v3EB_Nn045D7vmjh");
        //兩種寫法
        request.add(new MultiGetRequest.Item(
                "test_index",
                "MO57v3EB_Nn045D7aGgU"));
        MultiGetResponse response = client.mget(request, RequestOptions.DEFAULT);
        for (MultiGetItemResponse itemResponse : response) {
            Console.log(itemResponse.getResponse().getSourceAsString());
        }
    }
    
    @Test
    public void batchInsertData() throws IOException {
        //批量插入數據，更新和刪除同理
        BulkRequest request = new BulkRequest("test_index_" + DateUtil.format(new Date(), "yyyyMM"));
        for (int i = 0; i < 1000; i++) {
            TestData testData = new TestData();
            testData.setTestId(RandomUtil.randomLong(9999999999L));
            testData.setPrice(100.0D);
            testData.setName(RandomUtil.randomString(8));
            testData.setContent("");
            testData.setCreateTime(new Date());
            request.add(new IndexRequest().source(new JSONObject(testData, new JSONConfig().setDateFormat(DatePattern.NORM_DATETIME_PATTERN)).toString()
                    , XContentType.JSON));
        }
        BulkResponse response = client.bulk(request, RequestOptions.DEFAULT);
        Console.log("插入狀態:{} 數量:{} ",response.status(),response.getItems().length);
    }
    @Test
    public void updateByQuery() throws IOException {
        UpdateByQueryRequest request = new UpdateByQueryRequest("test_index");
        //默認情況下，版本沖突會中止 UpdateByQueryRequest 進程，但是你可以用以下命令來代替
        //設置版本沖突繼續
        request.setConflicts("proceed");
        //設置更新條件
        request.setQuery(QueryBuilders.rangeQuery("createTime").gte("2020-04-28 11:30:24").lte("2020-04-28 15:30:24"));
        //限制更新條數
        request.setMaxDocs(10);
        request.setScript(new Script(ScriptType.INLINE,"painless","ctx._source.testId=999999;", Collections.emptyMap()));
        BulkByScrollResponse response = client.updateByQuery(request, RequestOptions.DEFAULT);
        Console.log(response);
    }
    @Test
    public void query() throws IOException {
        SearchRequest request = new SearchRequest("test_index");
        SearchSourceBuilder builder = Criteria.create().addRangeQuery("createTime", "2020-04-28 11:30:24", "2020-04-28 15:30:24").builder();
        builder.from(0);
        builder.size(11);
        builder.sort("createTime", SortOrder.ASC);
        //不返回源數據。只有條數之類的數據。
//        builder.fetchSource(false);
        request.source(builder);
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);
        SearchHits hits = response.getHits();
        for (SearchHit hit : hits) {
            Console.log(hit.getSourceAsString());
        }
        Console.log("總數:{}",response.getHits().getTotalHits().value);
    }

    @After
    public void after() throws IOException {
        client.close();
    }
}

@Data
class TestData {
    private long testId;
    private double price;
    private String name;
    private String content;
    private Date createTime;
}

Criteria類

import cn.hutool.core.util.StrUtil;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import java.util.List;

/**
 * 復合查詢封裝
 */
public class Criteria {
    private final BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
    
    public static Criteria create(){
        return new Criteria();
    }
    
    /**
     * 條件增加完成后，獲取需要操作的對象
     * 
     * @return
     */
    public SearchSourceBuilder builder() {
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        return searchSourceBuilder.query(boolQueryBuilder);
    }

    /**
     * 增加條件查詢
     * 
     * @param fieldName
     * @param fieldValue
     */
    public Criteria addTermQuery(String fieldName, String fieldValue) {
        if (StrUtil.isNotBlank(fieldName) && StrUtil.isNotBlank(fieldValue)) {
            boolQueryBuilder.filter(QueryBuilders.termQuery(fieldName, fieldValue));
        }
        return this;
    }
    /**
     * 增加條件查詢
     * 主要針對 內容分詞后 精確匹配 fieldValue 不分詞
     * @param fieldName
     * @param fieldValue
     */
    public Criteria addMatchPhraseQuery(String fieldName, String fieldValue) {
        if (StrUtil.isNotBlank(fieldName) && StrUtil.isNotBlank(fieldValue)) {
            boolQueryBuilder.filter(QueryBuilders.matchPhraseQuery(fieldName, fieldValue));
        }
        return this;
    }

    /**
     * 增加區間查詢
     * 
     * @param fieldName
     * @param gteValue
     * @param lteValue
     */
    public Criteria addRangeQuery(String fieldName, Object gteValue, Object lteValue) {
        if (StrUtil.isNotBlank(fieldName)) {
            boolQueryBuilder.filter(QueryBuilders.rangeQuery(fieldName).gte(gteValue).lte(lteValue).includeLower(true)
                    .includeUpper(true));
        }
        return this;
    }

    /**
     * 增加包含查詢，相當於sql中的in
     * 
     * @param fieldName
     * @param values
     */
    public Criteria addTermsQuery(String fieldName, List<?> values) {
        if (StrUtil.isNotBlank(fieldName) && values != null && values.size() > 0) {
            boolQueryBuilder.filter(QueryBuilders.termsQuery(fieldName, values));
        }
        return this;
    }
    
    /**
     * 增加不包含查詢，相當於sql中的 not in
     * 
     * @param fieldName
     * @param values
     */
    public Criteria addNotTermsQuery(String fieldName, List<?> values) {
        if (StrUtil.isNotBlank(fieldName) && values != null && values.size() > 0) {
            boolQueryBuilder.mustNot(QueryBuilders.termsQuery(fieldName, values));
        }
        return this;
    }
    
}