微信公众号搜"智元新知"关注
微信扫一扫可直接关注哦！

Java Elasticsearch 使用

分类：Elasticsearch作者：编程之家原文地址

Java Elasticsearch 使用

概览

基于官方教程，提供Elasticsearch的基本操作，并编写单元测试：

  • 客户端连接
  • 客户端关闭
  • 建立索引
  • 删除索引
  • 插入文档
    • 单个文档的同步和异步插入
    • 多个文档的同步和异步插入
  • 查询文档

本地Docker的Elasticsearch搭建可参考：https://juejin.cn/post/6965321555475693582/

代码细节

Elasticsearch封装类

import lombok.SneakyThrows;
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.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AckNowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.*;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.sortBuilder;

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

/**
 * official document: https://www.elastic.co/guide/en/elasticsearch/client/java-rest/7.12/java-rest-high-document-bulk.html
 *
 * @author liuwei
 *
 * 自动重连说明：经过测试，在ES重启完成后，相关的操作便可恢复正常，无序额外操作
 */
public class Elasticclient {

    final private RestHighLevelClient client;

    /**
     * 构造Elasticsearch客户端
     * @param userName 用户名
     * @param password 密码
     * @param hosts 集群主机地址
     */
    public Elasticclient(final String userName, final String password, final HttpHost[] hosts) {
        UsernamePasswordCredentials credentials = new UsernamePasswordCredentials(userName, password);
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, credentials);

        client = new RestHighLevelClient(
                RestClient.builder(hosts)
                        .setHttpClientConfigCallback(builder -> builder.setDefaultCredentialsProvider(credentialsProvider)));
    }

    /**
     * 关闭客户端连接
     * @throws IOException IOException
     */
    public void close() throws IOException {
        client.close();
    }

    /**
     * 检测索引是否已建立
     * @param indexNames 索引名
     * @return bool
     */
    @SneakyThrows
    public boolean checkIndexExist(final String... indexNames) {
        GetIndexRequest request = new GetIndexRequest(indexNames);
        return client.indices().exists(request, RequestOptions.DEFAULT);
    }

    /**
     * 新建索引
     * @param indexName 索引名称
     * @param setting 索引配置内容（Json格式）
     * @return bool
     */
    @SneakyThrows
    public boolean createIndex(String indexName, String setting) {
        CreateIndexRequest request = new CreateIndexRequest(indexName);
        request.source(setting, XContentType.JSON);
        CreateIndexResponse createIndexResponse = client.indices().create(request, RequestOptions.DEFAULT);
        return createIndexResponse.isAckNowledged();
    }

    /**
     * 删除索引
     * @param indexName 删除索引
     * @return bool
     */
    @SneakyThrows
    public boolean deleteIndex(String indexName) {
        DeleteIndexRequest request = new DeleteIndexRequest(indexName);
        AckNowledgedResponse deleteIndexResponse = client.indices().delete(request, RequestOptions.DEFAULT);
        return deleteIndexResponse.isAckNowledged();
    }

    /**
     * 新增文档：同步阻塞
     * @param indexName 索引名
     * @param doc 文档
     * @param id 文档id
     * @return bool
     */
    @SneakyThrows
    public boolean addOneDocSync(String indexName, String doc, String id) {
        UpdateRequest request = new UpdateRequest(indexName, id).doc(doc, XContentType.JSON).upsert(doc, XContentType.JSON);
        UpdateResponse updateResponse = client.update(request, RequestOptions.DEFAULT);
        return updateResponse.status().equals(RestStatus.CREATED);
    }

    /**
     * 新增文档：异步非阻塞
     * @param indexName 索引名
     * @param doc 文档
     * @param id 文档id
     */
    @SneakyThrows
    public void addOneDocAsync(String indexName, String doc, String id) {
        ActionListener<UpdateResponse> listener = new ActionListener<UpdateResponse>() {
            @Override
            public void onResponse(UpdateResponse updateResponse) {

            }

            @Override
            public void onFailure(Exception e) {

            }
        };

        UpdateRequest request = new UpdateRequest(indexName, id).doc(doc, XContentType.JSON).upsert(doc, XContentType.JSON);
        client.updateAsync(request, RequestOptions.DEFAULT, listener);
    }

    /**
     * 批量新增文档：同步阻塞
     * @param indexName 索引名
     * @param docs 文档列表
     * @param ids ID列表
     * @return bool
     */
    @SneakyThrows
    public boolean addManyDocSync(String indexName, List<String> docs, List<String> ids) {
        if (docs.size() != ids.size()) {
            return false;
        }

        BulkRequest request = createBulkRequest(indexName, docs, ids);
        BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);
        return bulkResponse.status().equals(RestStatus.OK);

    }

    /**
     * 批量新增文档：异步非阻塞
     * @param indexName 索引名
     * @param docs 文档列表
     * @param ids ID列表
     * @return bool
     */
    @SneakyThrows
    public boolean addManyDocAsync(String indexName, List<String> docs, List<String> ids) {
        if (docs.size() != ids.size()) {
            return false;
        }

        ActionListener<BulkResponse> listener = new ActionListener<BulkResponse>() {
            @Override
            public void onResponse(BulkResponse bulkResponse) {

            }

            @Override
            public void onFailure(Exception e) {

            }
        };

        BulkRequest request = createBulkRequest(indexName, docs, ids);
        client.bulkAsync(request, RequestOptions.DEFAULT, listener);
        return true;
    }

    private BulkRequest createBulkRequest(String indexName, List<String> docs, List<String> ids) {
        BulkRequest request = new BulkRequest();
        for (int i = 0; i < docs.size(); i++) {
            String doc = docs.get(i);
            String id = ids.get(i);
            UpdateRequest r = new UpdateRequest(indexName, id).doc(doc, XContentType.JSON).upsert(doc, XContentType.JSON);
            request.add(r);
        }
        return request;
    }

    /**
     * 文档查询
     * @param pageIndex 分页：跳过的页数，从0开始
     * @param pageSize 分页：每页的数据量
     * @param queryCondition 查询条件
     * @param sortBuilders 排序条件
     * @param indexNames 索引名
     * @return {@link SearchHit} list
     */
    @SneakyThrows
    public SearchHit[] queryDoc(int pageIndex, int pageSize, MatchQueryBuilder queryCondition,
                                List<SortBuilder<?>> sortBuilders, String... indexNames) {
        SearchRequest searchRequest = new SearchRequest(indexNames);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(queryCondition);
        searchSourceBuilder.sort(sortBuilders);
        searchSourceBuilder.from(pageIndex * pageSize);
        searchSourceBuilder.size(pageSize);
        searchRequest.source(searchSourceBuilder);
        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        return searchResponse.getHits().getHits();
    }
}

测试类

junit5测试

import lombok.SneakyThrows;
import org.apache.commons.io.FileUtils;
import org.apache.http.HttpHost;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.sortOrder;
import org.junit.*;

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

@Ignore
public class ElasticclientTest {

    final private String index = "test_index";

    private Elasticclient elasticclient;

    @SneakyThrows
    @Test
    public void test() {
        testConnect();
        testAddindex();
        testAddOneDoc();
        testAddManyDoc();

        Thread.sleep(5000);
        testQuery();

        testReconnect();
        testDeleteIndex();
        testClose();
    }

    private void testConnect() {
        String host = "localhost";
        int port = 9200;
        String scheme = "http";
        final HttpHost[] hosts = new HttpHost[] {
                new HttpHost(host, port, scheme)
        };
        String userName = "elastic";
        String password = "BitWorker.2020";
        elasticclient = new Elasticclient(userName, password, hosts);

        if (elasticclient.checkIndexExist(index)) {
            assert elasticclient.deleteIndex(index);
        }
    }

    private void testAddindex() throws IOException {
        String indexSettingFilePath = "src/test/java/com/ninetech/cloud/support/elastic/TaskAndRobotLogIndex.json";
        File file=new File(indexSettingFilePath);
        String indexSetting = FileUtils.readFiletoString(file,"UTF-8");
        assert elasticclient.createIndex(index, indexSetting);
        assert elasticclient.checkIndexExist(index);
    }

    private void testAddOneDoc() {
        LogModel oneAsyncLog = createLogModel("taskIdOneAsync");
        elasticclient.addOneDocAsync(index, oneAsyncLog.toJsonString(), "taskIdOneAsync");

        LogModel onesyncLog = createLogModel("taskIdOnesync");
        assert elasticclient.addOneDocSync(index, onesyncLog.toJsonString(), "taskIdOnesync");
    }

    private void testAddManyDoc() {
        List<String> logModelsAsync = new ArrayList<>(2);
        logModelsAsync.add(createLogModel("taskIdManyAsync1").toJsonString());
        logModelsAsync.add(createLogModel("taskIdManyAsync2").toJsonString());

        List<String> idsAsync = new ArrayList<>(2);
        idsAsync.add("taskIdManyAsync1");
        idsAsync.add("taskIdManyAsync2");

        assert elasticclient.addManyDocAsync(index, logModelsAsync, idsAsync);

        List<String> logModelsSync = new ArrayList<>(2);
        logModelsSync.add(createLogModel("taskIdManySync1").toJsonString());
        logModelsSync.add(createLogModel("taskIdManySync2").toJsonString());

        List<String> idsSync = new ArrayList<>(2);
        idsSync.add("taskIdManySync1");
        idsSync.add("taskIdManySync2");

        assert elasticclient.addManyDocSync(index, logModelsSync, idsSync);
    }

    private LogModel createLogModel(String id) {
        return LogModel.builder()
                .message("test")
                .id(id)
                .createTime(new Date())
                .build();
    }

    private void testQuery() {
        MatchQueryBuilder queryCondition = new MatchQueryBuilder("id", "id");
        FieldSortBuilder createTimeSort = new FieldSortBuilder("createTime").order(SortOrder.ASC);
        SearchHit[] results = elasticclient.queryDoc(0, 100, queryCondition,
                Collections.singletonList(createTimeSort), index);
        Assert.assertEquals(0, results.length);

        queryCondition = new MatchQueryBuilder("message", "test");
        createTimeSort = new FieldSortBuilder("createTime").order(SortOrder.ASC);
        results = elasticclient.queryDoc(0, 100, queryCondition,
                Collections.singletonList(createTimeSort), index);
        Assert.assertEquals(6, results.length);
        for (SearchHit result : results) {
            System.out.println(result.toString());
        }

        queryCondition = new MatchQueryBuilder("id", "taskIdManySync1");
        createTimeSort = new FieldSortBuilder("createTime").order(SortOrder.ASC);
        results = elasticclient.queryDoc(0, 100, queryCondition,
                Collections.singletonList(createTimeSort), index);
        Assert.assertEquals(1, results.length);
        for (SearchHit result : results) {
            System.out.println(result.toString());
        }
    }

    private void testReconnect() throws InterruptedException {
        MatchQueryBuilder queryCondition = new MatchQueryBuilder("message", "test");
        FieldSortBuilder createTimeSort = new FieldSortBuilder("createTime").order(SortOrder.ASC);
        SearchHit[] results = elasticclient.queryDoc(0, 100, queryCondition,
                Collections.singletonList(createTimeSort), index);
        Assert.assertEquals(6, results.length);
        for (SearchHit result : results) {
            System.out.println(result.toString());
        }

        System.out.println("Waiting ES restart");
        Thread.sleep(1000 * 30);

        queryCondition = new MatchQueryBuilder("message", "test");
        createTimeSort = new FieldSortBuilder("createTime").order(SortOrder.ASC);
        results = elasticclient.queryDoc(0, 100, queryCondition,
                Collections.singletonList(createTimeSort), index);
        Assert.assertEquals(6, results.length);
        for (SearchHit result : results) {
            System.out.println(result.toString());
        }
    }

    private void testDeleteIndex() {
        assert elasticclient.deleteIndex(index);
        assert !elasticclient.checkIndexExist(index);
    }

    private void testClose() throws IOException {
        elasticclient.close();
    }
}

数据模型

import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;

import java.util.Date;

/**
 * @author lw1243925457
 */
@Builder
@Data
@AllArgsConstructor
public class LogModel {
    private String message;
    private String id;
    private Date createTime;

    public String toJsonString() {
        return JSONObject.toJSONString(this);
    }
}

索引配置文件：

{
  "mappings":{
    "properties":{
      "message":{
        "type":"text",
        "analyzer":"ik_max_word",
        "search_analyzer":"ik_smart"
      },
      "id":{
        "type":"keyword"
      },
      "createtime":{
        "type":"date",
        "format": "yyyy-MM-dd HH:mm:ss.SSS"
      }
    },
    "_source":{
      "enabled":true
    }
  },
  "settings":{
    "number_of_shards":2,
    "number_of_replicas":2
  }
}

参考链接

教程

问题

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。

相关推荐

完美掌握MySQL登录方法]MySQL登录教程
TCP/IP套接字登录方法是MySQL在一切服务平台都提供的一种登录方法，也是互联网中应用数最多的一种方法。这种方法在TCP/IP登录上创建一个根据互联网的登录申请，一般手机客户端在一台服务器上，而MySQL实例在另外一台网络服务器上，这两台设备根据TCP/IP协议登录。一般对于刚安裝好的mysql，在配置的时候勾选启用mysql lineclient的话，就可以直接通过“开始—mysql—mysql command lineclient”，点击进入以后，会提醒输入root登陆密码。...
easy-rules规则引擎最佳落地实践
easy-rule规则引擎最佳落地
windows 安装 Elasticsearch
Elasticsearch 是一个分布式、高扩展、高实时的搜索与数据分析引擎, 本文主要介绍了在 windows 10上 安装 Elasticsearch,Elasticsearch-head,kibana等软件, 方便自己学习使用,
关于ElasticSearch日期格式不一致的异常，可以这么解决
最近壹哥的一个学生，在利用spring-data-elasticsearch访问ElasticSearch(ES)时，遇到了一个关于日期类型的BUG，困扰了很久。然后他就找壹哥给他解决，接下来壹哥就把解决的过程给大家复现一下，希望本文可以给遇到同样问题的同学一点启发。我们先来看看他的POM.xml文件配置，如下所示：实体类的代码如下：创建索引的单元测试类代码如下：以下是ES中创建的索引 Mapping结构，其中日期类型的format为date_time。看了以上代码，你能猜出来哪里有问题吗
java 操作elasticsearch详细总结
java 操作elasticsearch详细总结
全文搜索引擎 Elasticsearch 入门教程
原文链接：http://www.ruanyifeng.com/blog/2017/08/elasticsearch.html全文搜索属于最常见的需求，开源的 Elasticsearch （以下简称 Elastic）是目前全文搜索引擎的首选。它可以快速地储存、搜索和分析海量数据。维基百科、Stack Overflow、Github 都采用它。Elastic 的底层是开源库 Lucene。但是，你没法...
elasticsearch启动常见错误
原文链接：https://www.cnblogs.com/zhi-leaf/p/8484337.html 问题出现环境，OS版本：CentOS-7-x86_64-Minimal-1708；ES版本：elasticsearch-6.2.2。1、max file descriptors [4096] for elasticsearch process is too low, increas...
【ES】Elasticsearch分词器使用和简单查询
文章目录Elasticsearch一. analysis与analyzer1.1 内置的分词器1.2 内置分词器示例1.3 中文分词1.3.1 IK分词器1.3.2 HanLP1.3.3 pinyin分词器1.4 中文分词演示1.5 分词的实际应用1.5.1 设置mapping1.5.2 插入数据1.5.3 查询1.6 拼音分词器1.6.1 设置settings1.6.2 设置mapping1.6.3 数据的插入1.6.4 查询1.7 中文、拼音混合查找1.7.1 设
Elasticsearch06：Elasticsearch分词详解：ES分词介绍、倒排索引介绍、分词器的作用、停用词
一、ES分词介绍ES中在添加数据，也就是创建索引的时候，会先对数据进行分词。在查询索引数据的时候，也会先根据查询的关键字进行分词。所以在ES中分词这个过程是非常重要的，涉及到查询的效率和准确度。假设有一条数据，数据中有一个字段是titile，这个字段的值为LexCorp BFG-9000。我们想要把这条数据在ES中创建索引，方便后期检索。创建索引和查询索引的大致流程是这样的：图中左侧是创建索引的过程：首先对数据进行空白字符分割，将LexCorp BFG-9000切分为LexCorp和BFG-90
ES和Kibana设置账号密码
命令执行后，会出现几句英文，大概的意思是：你如果确定开启密码的话，需要设置以下六种账户的密码elastic、apm_system、kibana、logstash_system、beats_system、remote_monitoring_user。第四步：重启后，输入kibana的访问地址：http://192.168.1.10:5601,这时已经提示让你输入账号密码了。第六步：重启后，输入ES的访问地址：http://192.168.1.10:9200，这时已经提示让你输入账号密码了。
苹果市值2025年有望达4万亿美元
pythonJavaScriptjavaHTMLreactjsC#AndroidCSSNode.jssqlrpython-3.xMysqLjQueryc++pandasFlutterangularIOSdjangolinuxswifttypescript路由器JSON路由器设置无线路由器h3c华三华三路由器设置华三路由器电脑软件教程arraysdocker软件图文教程Cvue.jslaravelspring-bootreact-native