ElasticSearch学习笔记二：使用Java客户端

一、前言

在上一篇文章中，我们对ES有了最基本的认识，本着实用为主的原则，我们先不学很深的东西，今天打算先学习一下ES的Java客户端如何使用。

二、创建项目

1、普通Maven项目

1、创建一个Maven项目

2、Pom文件

<dependencies>

  <!--ES客户端-->
  <dependency>
    <groupId>co.elastic.clients</groupId>
    <artifactId>elasticsearch-java</artifactId>
    <version>7.17.25</version>
  </dependency>

  <!--JSON序列化-->
  <dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.17.0</version>
  </dependency>

  <!--lombok:用于生成GET/SET 简化开发-->
  <dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <version>1.18.30</version>
  </dependency>

</dependencies>

3、Coding

（1）创建ES客户端

 /**
     * 获取ES客户端
     * @return es Java客户端
     */
    private static ElasticsearchClient getEsClient() {
        //Rest客户端，可以理解为是一个Http客户端，用于发送http请求
        RestClient restClient = RestClient.builder(new HttpHost("localhost", 9200)).build();
        //ElasticsearchTransport用于和ES集群通信，封装了各种方法，第二个参数则是设置序列化方式
        ElasticsearchTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
        return new ElasticsearchClient(transport);
    }

（2）判断索引Product是否存在，如果不存在则创建索引。（当然通常情况下创建索引的操作是手动操作的，就类似创建数据表）

 /**
     * 校验并创建索引，如果存在直接返回true
     * 如果不存在则创建索引，同时返回是否创建成功的结果
     */
    private static boolean checkAndCreateIndex(final ElasticsearchIndicesClient indices) throws IOException {
        //构建索引是否存在的请求参数
        ExistsRequest existsRequest = new ExistsRequest.Builder().index("product").build();
        final BooleanResponse exists = indices.exists(existsRequest);
        if (exists.value()) {
            System.out.println("索引已经存在,不用再创建了");
            return true;
        }
        //Java17的新特性（这样写字符串真的很方便）
        Reader createIndexJson = new StringReader("""
            {
              "mappings": {
                "properties": {
                  "id":{
                    "type": "long"
                  },
                  "name":{
                    "type": "text",
                    "analyzer":"ik_max_word"
                  },
                  "price":{
                    "type": "double"
                  }
                }
              }
            }""");
        //创建索引
        CreateIndexRequest createIndexRequest = new CreateIndexRequest.Builder().index("product") //索引名
            .includeTypeName(false) //是否包含包名
            .settings(new IndexSettings.Builder().numberOfShards("1").numberOfReplicas("1").build())
            .withJson(createIndexJson).build();
        final CreateIndexResponse createIndexResponse = indices.create(createIndexRequest);
        System.out.println("创建索引是否成功：" + createIndexResponse.acknowledged());
        return createIndexResponse.acknowledged();
    }

（3）批量写入数据

    /**
     * 批量写入数据
     */
    private static boolean bulkWriteDoc(final ElasticsearchClient esClient) throws IOException {
        final List<Product> products = generalProduct(100);

        //批量写入
        BulkRequest.Builder br = new BulkRequest.Builder();
        for (Product product : products) {
            br.operations(op -> op.index(idx -> idx.index("product").id(product.getId().toString()).document(product)));
        }
        BulkResponse bulkResponse = esClient.bulk(br.build());
        System.out.println("批量写入结果是否成功：" + !bulkResponse.errors());
        return !bulkResponse.errors();
    }


//product的代码

@Data
@NoArgsConstructor
@AllArgsConstructor
public class Product {

    private Long id;

    private String name;

    private Double price;
}

（4）查询数据

//根据ID查询
GetResponse<Product> response = esClient.get(g -> g.index("product").id("1"), Product.class);
if (response.found()) {
    System.out.println("根据ID查询到对应的数据 " + response.source());
} else {
    System.out.println("根据ID查询未对应的数据");
}

//根据条件查询：例如搜索名称为商品20的数据
SearchResponse<Product> queryResponse = esClient.search(
    s -> s.index("product").query(q -> q.match(t -> t.field("name").query("商品20"))), Product.class);
TotalHits total = queryResponse.hits().total();
assert total != null;
boolean isExactResult = total.relation() == TotalHitsRelation.Eq;

if (isExactResult) {
    System.out.println("命中的文档数量为：" + total.value());
} else {
    System.out.println("没有命中任务数据");
}

List<Hit<Product>> hits = queryResponse.hits().hits();
for (Hit<Product> hit : hits) {
    Product product = hit.source();
    System.out.println("命中的数据：" + product);
}

（5）完整代码

package com.cmxy.esdemo;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.GetResponse;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.TotalHits;
import co.elastic.clients.elasticsearch.core.search.TotalHitsRelation;
import co.elastic.clients.elasticsearch.indices.CreateIndexRequest;
import co.elastic.clients.elasticsearch.indices.CreateIndexResponse;
import co.elastic.clients.elasticsearch.indices.ElasticsearchIndicesClient;
import co.elastic.clients.elasticsearch.indices.ExistsRequest;
import co.elastic.clients.elasticsearch.indices.IndexSettings;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.endpoints.BooleanResponse;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import com.cmxy.entity.Product;
import java.io.IOException;
import java.io.Reader;
import java.io.StringReader;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.ArrayList;
import java.util.List;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;

public class ESTest {

    public static void main(String[] args) throws IOException, InterruptedException {
        try (ElasticsearchClient esClient = getEsClient()) {
            //获取es客户端
            //判断索引是否存在
            final ElasticsearchIndicesClient indices = esClient.indices();
            //判断索引是否存在，如果不存在则创建
            boolean createIndexSuccess = checkAndCreateIndex(indices);

            //往索引里写入数据
            boolean writeSuccess = false;
            if (createIndexSuccess) {
                writeSuccess = bulkWriteDoc(esClient);
            }

            //写入成功后，查询
            if (writeSuccess) {
                queryData(esClient);
            }
        }

    }

    private static void queryData(final ElasticsearchClient esClient) throws InterruptedException, IOException {
        //阻塞一下，否则刚写入直接查询会查不到数据
        Thread.sleep(2000L);
        //根据ID查询
        GetResponse<Product> response = esClient.get(g -> g.index("product").id("1"), Product.class);
        if (response.found()) {
            System.out.println("根据ID查询到对应的数据 " + response.source());
        } else {
            System.out.println("根据ID查询未对应的数据");
        }

        //根据条件查询：例如搜索名称为商品20的数据
        SearchResponse<Product> queryResponse = esClient.search(
            s -> s.index("product").query(q -> q.match(t -> t.field("name").query("商品20"))), Product.class);
        TotalHits total = queryResponse.hits().total();
        assert total != null;
        boolean isExactResult = total.relation() == TotalHitsRelation.Eq;

        if (isExactResult) {
            System.out.println("命中的文档数量为：" + total.value());
        } else {
            System.out.println("没有命中任务数据");
        }

        List<Hit<Product>> hits = queryResponse.hits().hits();
        for (Hit<Product> hit : hits) {
            Product product = hit.source();
            System.out.println("命中的数据：" + product);
        }
    }

    /**
     * 批量写入数据
     */
    private static boolean bulkWriteDoc(final ElasticsearchClient esClient) throws IOException {
        final List<Product> products = generalProduct(100);

        //批量写入
        BulkRequest.Builder br = new BulkRequest.Builder();
        for (Product product : products) {
            br.operations(op -> op.index(idx -> idx.index("product").id(product.getId().toString()).document(product)));
        }
        BulkResponse bulkResponse = esClient.bulk(br.build());
        System.out.println("批量写入结果是否成功：" + !bulkResponse.errors());
        return !bulkResponse.errors();
    }

    /**
     * 校验并创建索引，如果存在直接返回true
     * 如果不存在则创建索引，同时返回是否创建成功的结果
     */
    private static boolean checkAndCreateIndex(final ElasticsearchIndicesClient indices) throws IOException {
        //构建索引是否存在的请求参数
        ExistsRequest existsRequest = new ExistsRequest.Builder().index("product").build();
        final BooleanResponse exists = indices.exists(existsRequest);
        if (exists.value()) {
            System.out.println("索引已经存在,不用再创建了");
            return true;
        }
        //Java17的新特性（这样写字符串真的很方便）
        Reader createIndexJson = new StringReader("""
            {
              "mappings": {
                "properties": {
                  "id":{
                    "type": "long"
                  },
                  "name":{
                    "type": "text",
                    "analyzer":"ik_max_word"
                  },
                  "price":{
                    "type": "double"
                  }
                }
              }
            }""");
        //创建索引
        CreateIndexRequest createIndexRequest = new CreateIndexRequest.Builder().index("product") //索引名
            .includeTypeName(false) //是否包含包名
            .settings(new IndexSettings.Builder().numberOfShards("1").numberOfReplicas("1").build())
            .withJson(createIndexJson).build();
        final CreateIndexResponse createIndexResponse = indices.create(createIndexRequest);
        System.out.println("创建索引是否成功：" + createIndexResponse.acknowledged());
        return createIndexResponse.acknowledged();
    }

    private static List<Product> generalProduct(int count) {
        List<Product> products = new ArrayList<>();
        for (int i = 1; i <= count; i++) {
            products.add(new Product((long) i, "商品" + i,
                BigDecimal.valueOf(Math.random() * 1000).setScale(2, RoundingMode.HALF_UP).doubleValue()));
        }
        return products;
    }

    /**
     * 获取ES客户端
     *
     * @return es Java客户端
     */
    private static ElasticsearchClient getEsClient() {
        //Rest客户端，可以理解为是一个Http客户端，用于发送http请求
        RestClient restClient = RestClient.builder(new HttpHost("localhost", 9200)).build();
        //ElasticsearchTransport用于和ES集群通信，封装了各种方法，第二个参数则是设置序列化方式
        ElasticsearchTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
        return new ElasticsearchClient(transport);
    }
}

(6)运行结果

这里可能有的朋友会有点疑惑包括笔者一开始也是，为什么我搜索的是“商品20”，能命中的文档数居然是100，按理说不应该只有一条吗？还有为什么命中了100条只返回了10条呢？

其实是这样的，因为我们当前的product索引，他的name是一个text类型，是会被分词的，我们可以看下他分词后是涨什么样子的

在kibana中执行如下命令

POST /product/_analyze
{
  "field": "name",
  "text": "商品20"
}

结果：
{
  "tokens" : [
    {
      "token" : "商品",
      "start_offset" : 0,
      "end_offset" : 2,
      "type" : "CN_WORD",
      "position" : 0
    },
    {
      "token" : "20",
      "start_offset" : 2,
      "end_offset" : 4,
      "type" : "ARABIC",
      "position" : 1
    }
  ]
}

我们可以看到对于name的分词，分为“商品”和“20”两个词，且match时默认是用or，换成我们熟悉的Mysql，那就是类似于 select * from product where namelike ‘%商品%’ or name like ‘%20%’，所以所有的的数据就查到了。

例如我插入了一个产品，名称为测试20，我再执行查询语句：

GET /product/_search
{
  "size": 200, 

  "sort": [
    {
      "id": {
        "order": "desc"
      }
    }
  ], 
 "query": {
   "match": {
     "name": {
       "query": "商品20",
       "analyzer": "ik_max_word"
     }
   }
 }
}

结果如下图

2、Springboot整合ESClient

2.1、使用ES原生客户端

(1)Pom文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.cmxy</groupId>
  <artifactId>springboot-es</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <name>springboot-es</name>
  <description>springboot-es</description>
  <properties>
    <java.version>1.8</java.version>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    <spring-boot.version>2.6.13</spring-boot.version>
    <jakarta.json>2.0.1</jakarta.json>
  </properties>
  <dependencies>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-test</artifactId>
      <scope>test</scope>
    </dependency>

    <dependency>
      <groupId>org.elasticsearch</groupId>
      <artifactId>elasticsearch</artifactId>
      <version>7.17.25</version>
    </dependency>
    <dependency>
      <groupId>co.elastic.clients</groupId>
      <artifactId>elasticsearch-java</artifactId>
      <version>7.17.25</version>
    </dependency>
    <!-- 覆盖springboot维护的版本 -->
    <dependency>
      <groupId>org.elasticsearch.client</groupId>
      <artifactId>elasticsearch-rest-client</artifactId>
      <version>7.17.25</version>
    </dependency>
    <dependency>
      <groupId>com.fasterxml.jackson.core</groupId>
      <artifactId>jackson-databind</artifactId>
      <version>2.12.3</version>
    </dependency>
    <dependency>
      <groupId>jakarta.json</groupId>
      <artifactId>jakarta.json-api</artifactId>
      <version>2.0.1</version>
    </dependency>
    <dependency>
      <groupId>org.projectlombok</groupId>
      <artifactId>lombok</artifactId>
    </dependency>

  </dependencies>
  <dependencyManagement>
    <dependencies>
      <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-dependencies</artifactId>
        <version>${spring-boot.version}</version>
        <type>pom</type>
        <scope>import</scope>
      </dependency>
    </dependencies>
  </dependencyManagement>

  <build>
    <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <version>3.8.1</version>
        <configuration>
          <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>${spring-boot.version}</version>
                <configuration>
                    <mainClass>com.cmxy.springbootes.SpringbootEsApplication</mainClass>
                    <skip>true</skip>
                </configuration>
                <executions>
                    <execution>
                        <id>repackage</id>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

（2）将EsClient注入Spring容器

@Component
public class EsConfig {

    @Bean
    public ElasticsearchClient esClient() {
        // Create the low-level client
        RestClient restClient = RestClient.builder(
            new HttpHost("localhost", 9200)).build();
        // Create the transport with a Jackson mapper
        ElasticsearchTransport transport = new RestClientTransport(
            restClient, new JacksonJsonpMapper());

        // And create the API client
        return new ElasticsearchClient(transport);
    }
}

(3)在要用的地方引入

@RestController
@RequestMapping("/es")
public class EsController {

    @Resource
    private ElasticsearchClient elasticsearchClient;

    @GetMapping("/testEs")
    public boolean testEs() throws IOException {
        ExistsRequest request = new ExistsRequest.Builder()
        .index("product")
        .build();
        BooleanResponse exists = elasticsearchClient.indices().exists(request);
        return exists.value();
    }

2.2、使用springData

（1）添加依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.cmxy</groupId>
    <artifactId>springboot-es</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>springboot-es</name>
    <description>springboot-es</description>
    <properties>
        <java.version>1.8</java.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <spring-boot.version>2.6.13</spring-boot.version>
        <jakarta.json>2.0.1</jakarta.json>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
        </dependency>



    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>

</dependencies>
<dependencyManagement>
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-dependencies</artifactId>
        <version>${spring-boot.version}</version>
        <type>pom</type>
        <scope>import</scope>
    </dependency>
</dependencies>
</dependencyManagement>

<build>
<plugins>
    <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <version>3.8.1</version>
        <configuration>
            <source>1.8</source>
            <target>1.8</target>
            <encoding>UTF-8</encoding>
        </configuration>
    </plugin>
    <plugin>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-maven-plugin</artifactId>
        <version>${spring-boot.version}</version>
        <configuration>
            <mainClass>com.cmxy.springbootes.SpringbootEsApplication</mainClass>
            <skip>true</skip>
        </configuration>
        <executions>
            <execution>
                <id>repackage</id>
                <goals>
                    <goal>repackage</goal>
                </goals>
            </execution>
        </executions>
    </plugin>
</plugins>
</build>

</project>

（2）添加配置文件

spring:
  elasticsearch:
    uris: localhost:9200

（3）编写Repository

package com.cmxy.springbootes.demos.repository;

import com.cmxy.springbootes.demos.entity.Product;
import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
import org.springframework.stereotype.Repository;

@Repository
public interface ProductEsRepository extends ElasticsearchRepository<Product,Long> {
}

（4）Controller

package com.cmxy.springbootes.demos.service;

import com.cmxy.springbootes.demos.entity.Product;
import com.cmxy.springbootes.demos.repository.ProductEsRepository;
import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.SearchOperations;
import org.springframework.data.elasticsearch.core.query.Criteria;
import org.springframework.data.elasticsearch.core.query.CriteriaQuery;
import org.springframework.data.elasticsearch.core.query.Query;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

@RestController
@RequestMapping("/es")
public class EsController {

    @Resource
    private ElasticsearchOperations elasticsearchOperations;
    @Resource
    private SearchOperations searchOperations;

    @Resource
    private ProductEsRepository productEsRepository;

    /**
     * 校验索引是否存在
     *
     * @return
     */
    @GetMapping("/checkIndex")
    public boolean checkIndexExists() {
        return elasticsearchOperations.indexOps(Product.class).exists();
    }

    /**
     * 创建索引
     */
    @PostMapping("/createIndex")
    public boolean createIndex() {
        return elasticsearchOperations.indexOps(Product.class).create();
    }

    /**
     * 批量写入文档
     */
    @PostMapping("/batchCreateDocument")
    public boolean batchCreateDocument() {
        List<Product> products = new ArrayList<>();
        for (int i = 1; i <= 100; i++) {
            products.add(new Product((long) i, "商品" + i, BigDecimal.valueOf(Math.random() * 1000).setScale(2, RoundingMode.HALF_UP).doubleValue()));
        }
        productEsRepository.saveAll(products);
        return true;
    }

    /**
     * 根据ID查询
     *
     * @param id
     * @return
     */
    @GetMapping("/getById")
    public Product getById(Long id) {
        //当然也可以这几使用searchOperations操作，类似 repository和mapper的关系
        Product product = productEsRepository.findById(id).orElse(null);
        System.out.println(product);
        return product;
    }


    @GetMapping("/query")
    public List<Product> query() {
        Criteria criteria = new Criteria("name").is("商品20");
        Query query = new CriteriaQuery(criteria);
        SearchHits<Product> searchHits = searchOperations.search(query, Product.class);
        if (searchHits.hasSearchHits()) {
            List<SearchHit<Product>> hits = searchHits.getSearchHits();
            return hits.stream().map(SearchHit::getContent).collect(Collectors.toList());
        }
        return null;
    }

}