[大数据]Elasticsearch的原理、安装、使用与封装好的Java工具类

Elasticsearch介绍

Elasticsearch是一个使用Java语言并且基于Lucene编写的搜索引擎框架，它提供了分布式的全文搜索功能，还提供了统一的基于RESTful风格的web接口，客户端也提供了相应的API。

倒排索引

Elasticsearch将存储的数据以一定的方式进行分词，然后将分词的内容存放到一个分词库中，当用户查询数据的时候，会先将用户输入的关键字进行分词，然后去分词库中匹配内容，最终得到数据的标识，最后拿着这个标识，去存放数据的位置，找到指定的数据。

Elasticsearch的结构

• Elasticsearch的服务中，可以创建多个索引。
• 每一个索引默认被分成5片存储。
• 每一个分片都会存在至少一个备份分片。
• 备份分片默认不会帮助检索数据，当Elasticsearch检索压力特别大的时候，备份分片才会帮助检索数据。
• 备份的分片必须放在不同的服务器中。

一个服务包含多个索引，一个索引包含多个文档，一个文档包含多个属性

类似于：一个数据库包含多个表，一个表包含多行，一行包含多列

安装Elasticsearch和Kibana

1、创建docker-compose.yml文件

（environment中需要替换为自己的主机IP，image可以替换为指定的镜像）

version: "3.1"
services:
  elasticsearch:
    image: daocloud.io/library/elasticsearch:7.6.2
    restart: always
    container_name: elasticsearch
    ports:
      - 9200:9200
    environment:
      - discovery.type=single-node
  kibana:
    image: daocloud.io/library/kibana:7.6.2
    restart: always
    container_name: kibana
    ports:
      - 5601:5601
    environment:
      - elasticsearch_url=http://192.168.229.128:9200
    depends_on:
      - elasticsearch

2、修改虚拟机最大内存

vi /etc/sysctl.conf
# 在最后一行添加
vm.max_map_count=655360

# 保存后重启
sysctl -p

3、查看elasticsearch容器的IP

# 将查看到的IP记住
docker inspect elasticsearch

4、进入kibana容器内部，修改

docker exec -it kibana bash
vi config/kibana.yml

# 将这个换成刚刚查询到的elasticsearch的IP和端口
elasticsearch.hosts: [ "http://172.18.0.2:9200" ]
# 将这个改为false
xpack.monitoring.ui.container.elasticsearch.enabled: false

最后退出容器

5、重启docker

docker-compose restart

不要docker-compose down，因为down之后，刚刚在kibana容器内部的配置就都重新刷新了

kibana是一个Elasticsearch的图形化界面

安装IK分词器

进入es容器，进入到bin目录，执行以下命令

elasticsearch-plugin install https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v7.6.2/elasticsearch-analysis-ik-7.6.2.zip

更多版本

https://github.com/medcl/elasticsearch-analysis-ik/releases

Elasticsearch简单操作

创建一个索引

PUT /book
{
  "settings": {
    "number_of_shards": 5,
    "number_of_replicas": 1
  },
  "mappings": {
    "properties": {
      "name": {
        "type": "text"
      },
      "author": {
        "type": "keyword"
      },
      "price": {
        "type": "double" 
      },
      "createTime": {
        "type": "date",
        "format": "yyyy-MM-dd HH:mm:ss || yyyy-MM-dd || epoch_millis"
      }
    }
  }
}

删除索引

DELETE book

创建文档

PUT /book/_doc/1
{
  "name": "深入理解Java虚拟机",
  "author": "张某某",
  "price": "99.99",
  "createTime": "2021-07-10 15:15:00"
}

查询索引

GET book/_doc/1

Java操作Elasticsearch的工具类

1、导入依赖

<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch</artifactId>
    <version>7.6.2</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>
<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
    <version>7.6.2</version>
</dependency>

2、编写配置文件

elasticsearch:
  host: 39.102.58.134
  port: 9200
  connTimeout: 3000
  socketTimeout: 5000
  connectionRequestTimeout: 500

3、创建Elasticsearch配置类

package com.robot.poem.config;

import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Elasticsearch配置类。
 *
 * @Author 张宝旭
 * @Date 2021/2/20
 */
@Configuration
public class ElasticsearchConfig {

    @Value("${elasticsearch.host}")
    private String host;

    @Value("${elasticsearch.port}")
    private int port;

    @Value("${elasticsearch.connTimeout}")
    private int connTimeout;

    @Value("${elasticsearch.socketTimeout}")
    private int socketTimeout;

    @Value("${elasticsearch.connectionRequestTimeout}")
    private int connectionRequestTimeout;

    /**
     * 获取连接。
     *
     * @return RestHighLevelClient
     */
    @Bean
    public RestHighLevelClient initRestClient() {
        RestClientBuilder builder = RestClient.builder(new HttpHost(host, port))
                .setRequestConfigCallback(requestConfigBuilder -> requestConfigBuilder
                        .setConnectTimeout(connTimeout)
                        .setSocketTimeout(socketTimeout)
                        .setConnectionRequestTimeout(connectionRequestTimeout));
        return new RestHighLevelClient(builder);
    }
}

4、编写Elasticsearch工具类

package com.robot.poem.utils;

import com.alibaba.fastjson.JSON;
import com.robot.poem.config.ElasticsearchConfig;
import com.robot.poem.pojo.Poem;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.List;
import java.util.Map;

/**
 * elasticsearch。
 *
 * @Author 张宝旭
 * @Date 2020/11/8
 */
@Component
public class ElasticsearchUtils {

    @Resource
    private ElasticsearchConfig elasticsearchConfig;

    // 索引名称
    private static String indexName = "poem-list";

    // 创建索引
    public boolean createIndex() throws IOException {

        Settings.Builder settings = Settings.builder()
                .put("number_of_shards", 5)
                .put("number_of_replicas", 1);

        XContentBuilder mapping = JsonXContent.contentBuilder()
                .startObject()
                .startObject("properties")
                .startObject("poemId")
                .field("type", "long")
                .endObject()
                .startObject("title")
                .field("type", "text")
                .field("analyzer", "ik_max_word")
                .endObject()
                .startObject("author")
                .field("type", "text")
                .field("analyzer", "ik_max_word")
                .endObject()
                .startObject("dynasty")
                .field("type", "text")
                .field("analyzer", "ik_max_word")
                .endObject()
                .startObject("content")
                .field("type", "text")
                .field("analyzer", "ik_max_word")
                .endObject()
                .startObject("cover")
                .field("type", "keyword")
                .endObject()
                .startObject("likes")
                .field("type", "integer")
                .endObject()
                .startObject("collect")
                .field("type", "integer")
                .endObject()
                .startObject("read")
                .field("type", "integer")
                .endObject()
                .endObject()
                .endObject();

        // 添加索引
        CreateIndexRequest request = new CreateIndexRequest(indexName);

        request.settings(settings);
        request.mapping(mapping);

        CreateIndexResponse response = elasticsearchConfig.initRestClient().indices().create(request, RequestOptions.DEFAULT);

        return response.isAcknowledged();
    }

    // 添加多条数据
    public boolean addIndexDataList(List<Poem> poemList) throws IOException {
        BulkRequest request = new BulkRequest();
        for (Poem poem : poemList) {
            request.add(new IndexRequest(indexName).id(String.valueOf(poem.getPoemId())).source(JSON.toJSONString(poem), XContentType.JSON));
        }
        BulkResponse response = elasticsearchConfig.initRestClient().bulk(request, RequestOptions.DEFAULT);
        return response.status().equals("OK");
    }

    // 添加单条数据
    public boolean addIndexData(Poem poem) throws IOException {
        BulkRequest request = new BulkRequest();
        request.add(new IndexRequest(indexName).id(String.valueOf(poem.getPoemId())).source(JSON.toJSONString(poem), XContentType.JSON));
        BulkResponse response = elasticsearchConfig.initRestClient().bulk(request, RequestOptions.DEFAULT);
        return response.status().equals("OK");
    }

    // 查询数据
    public SearchHit[] queryData(String field, String fieldValue, int from, int size) throws IOException {
        // 创建request对象
        SearchRequest request = new SearchRequest(indexName);

        // 封装request对象
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        from = from <= -1 ? 0 : from;
        size = Math.min(size, 1000);
        size = size <= 0 ? 10 : size;
        // 开始位置
        sourceBuilder.from(from);
        // 每页大小
        sourceBuilder.size(size);
        sourceBuilder.query(QueryBuilders.matchQuery(field, fieldValue));

        // 发送request
        request.source(sourceBuilder);
        SearchResponse response = elasticsearchConfig.initRestClient().search(request, RequestOptions.DEFAULT);
//        System.out.println("ES查询: " + response.status());

        return response.getHits().getHits();
    }

    // 查询数据
    public Map<String, Object> queryDataById(String id) throws IOException {
        // 创建request对象
        SearchRequest request = new SearchRequest(indexName);

        // 封装request对象
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(QueryBuilders.termQuery("_doc", id));

        // 发送request
        SearchResponse response = elasticsearchConfig.initRestClient().search(request, RequestOptions.DEFAULT);
        request.source(sourceBuilder);
//        System.out.println("ES查询: " + response.status());

        //        Map<String, Object> sourceAsMap = hits[0].getSourceAsMap();
        Map<String, Object> map = null;
        for (SearchHit hit : response.getHits().getHits()) {
            map = hit.getSourceAsMap();
            break;
        }
        return map;
    }

    // 删除索引中指定ID的一条数据
    public void deleteData(String id) throws IOException {
        DeleteRequest request = new DeleteRequest(indexName, id);
        DeleteResponse response = elasticsearchConfig.initRestClient().delete(request, RequestOptions.DEFAULT);
        System.out.println("ES删除: " + response.status());
    }

    // 修改
    public boolean updateData(Poem poem) throws IOException {
        Map map = JSON.parseObject(JSON.toJSONString(poem), Map.class);
        UpdateRequest request = new UpdateRequest(indexName, String.valueOf(map.get("id")));
        request.doc(map);
        UpdateResponse response = elasticsearchConfig.initRestClient().update(request, RequestOptions.DEFAULT);
        return response.status().equals("OK");
    }

}

在创建索引的代码中，startObject就是一个文档的开始，endObject就是结束，中间可以填加一些条件

例如以下，为poemId字段添加索引文档，条件为poemId类型是long类型

            .startObject("poemId")
            .field("type", "long")
            .endObject()

其中.startObject(“properties”)这一层里面包含的就是全部的属性，也就是一个索引包含所有文档

下面的number_of_shards为索引的片数，number_of_replicas为索引的备份数量

    Settings.Builder settings = Settings.builder()
            .put("number_of_shards", 5)
            .put("number_of_replicas", 1);