Elasticsearch 使用 Java High Level

前边我们学习了如何使用 RESTful API 去操作 ES，这种方式可能在实际项目中用的比较少，但这些内容都是必备的基础，对后续的学习还是很有帮助的，还是需要掌握的。

Elasticsearch Clients 提供了许多语言的支持，我们要学习其中的 Java REST Client，通过编写 Java 代码的方式来操作 ES。

其中 Java REST Client 又分为 Java Low Level REST Client 和 Java High Level REST Client，我们要使用的是 Java High Level REST Client：

注意，Java High Level REST Client 最低需要 Java1.8 的版本。

1、添加依赖

这里我们创建一个 Spring Boot 项目，添加如下依赖来引入 Java High Level REST Client：

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

由于 ES 的版本更新比较快，基本每个月会更新1-2个版本，但 Spring Data Elasticsearch 对 ES 新版本的支持还是比较滞后的，如果 Spring Data Elasticsearch 对应的 ES 版本比你安装的 ES 版本低，建议直接修改 Spring Data Elasticsearch 对应的 ES 版本，使其和你安装的 ES 版本保持一致。

我创建的基于 Spring Boot 2.3.6.RELEASE 的项目，其中的 ES 版本为7.6.2：

我之前安装的是 ES7.9.3 版本，所以可通过如下方式修改 Spring Data Elasticsearch 对应的 ES 版本号（目前最新的 Spring Boot 2.4.0 已经支持 ES7.9.3 了）：

<properties>
    <elasticsearch.version>7.9.3</elasticsearch.version>
</properties>

通过 Spring Data Elasticsearch 来集成 Java High Level REST Client，就可以使用 Spring Data 的一些特性来简化开发，但前期我们先学习 Java High Level REST Client 的原生 API，这个更具有通用性，掌握了原生 API 再学习整合 Spring Data 后的一些特性就很容易了。

由于还会用到 JSON 相关的操作，这里添加fastjson依赖：

<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.2.73</version>
</dependency>

2、初始化

创建如下配置类，来连接到 ES 节点，创建RestHighLevelClient对象，这样初始化工作就完成了。RestHighLevelClient是重点，后续的各种操作都要使用它：

@Configuration
public class ElasticsearchConfig {
    @Bean
    public RestHighLevelClient restHighLevelClient() {
        RestHighLevelClient client = new RestHighLevelClient(
                RestClient.builder(
                        new HttpHost("127.0.0.1", 9200, "http"),
                        new HttpHost("127.0.0.1", 9201, "http"),
                        new HttpHost("127.0.0.1", 9202, "http")));
        return client;
    }
}

我们先创建一个UserService类，里边注入通过配置类生成的RestHighLevelClient对象，所有的操作的方法都在该类里完成：

@Service
public class UserService {
    @Autowired
    @Qualifier("restHighLevelClient")
    private RestHighLevelClient client;
}

3、创建索引

如下的代码会创建一个名为user的索引：

public void createIndex() throws IOException {
    CreateIndexRequest request = new CreateIndexRequest("user");
    CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);
    System.out.println(response.isAcknowledged());
}

创建索引时也可以根据需要指定一些配置信息，例如分片数量、文档字段的映射信息、索引别名等：

public void createIndex2() throws IOException {
    CreateIndexRequest request = new CreateIndexRequest("user");

    // 索引分片数量配置
    request.settings(Settings.builder()
            .put("index.number_of_shards", 2)
            .put("index.number_of_replicas", 1));

    // 设置文档字段的映射信息
    Map<String, Object> birthday = new HashMap<>();
    birthday.put("type", "date");
    birthday.put("format", "yyyy-MM-dd");
    Map<String, Object> properties = new HashMap<>();
    properties.put("birthday", birthday);
    Map<String, Object> mapping = new HashMap<>();
    mapping.put("properties", properties);
    request.mapping(mapping);
        
    // 通过json设置文档字段的映射信息
//    request.mapping("{\n" +
//            "   \"properties\": {\n" +
//            "       \"birthday\": {\n" +
//            "           \"type\": \"date\",\n" +
//            "           \"format\": \"yyyy-MM-dd\"\n" +
//            "       }\n" +
//            "   }\n" +
//            "}", XContentType.JSON);

    // 设置索引别名
    request.alias(new Alias("user_alias"));

    CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);
    System.out.println(response.isAcknowledged());
}

3、删除索引

删除索引前可以先判断索引是否存在：

public boolean existsIndex() throws IOException {
    GetIndexRequest request = new GetIndexRequest("user");
    boolean exists = client.indices().exists(request, RequestOptions.DEFAULT);
    return exists;
}

然后再删除：

public void deleteIndex() throws IOException {
    if (!existsIndex()) {
        return;
    }
    DeleteIndexRequest request = new DeleteIndexRequest("user");
    AcknowledgedResponse response = client.indices().delete(request, RequestOptions.DEFAULT);
    System.out.println(response.isAcknowledged());
}

4、添加文档

创建好了索引就可以给里边添加文档数据了，首先看添加单个文档：

public void addDocument() throws IOException {
    User user = new User();
    user.setName("张三");
    user.setAge(30);
    user.setBirthday("1990-03-12");
    user.setSchool("清华");

    IndexRequest request = new IndexRequest("user");
//    request.timeout(TimeValue.timeValueSeconds(2));
    // 超时时间
    request.timeout("2s");
    // 文档id
    request.id("1");
    // 设置要添加的数据
    request.source(JSON.toJSONString(user), XContentType.JSON);
    IndexResponse response = client.index(request, RequestOptions.DEFAULT);
    System.out.println(response.status());
}

我们这里以将对象转成 JSON 串再添加，这种相对简单通用些。添加时如果不提供文档 id，ES 会给一个默认值，这里为了方便后边演示就指定了文档 id。

官方还提供了其它方式，可以查看文档。

如果需要批量添加我们可以使用BulkRequest类来完成，具体的实现如下：

public void bulkAddDocument() throws IOException {
        User user1 = new User();
        user1.setName("李四");
        user1.setAge(18);
        user1.setBirthday("2002-01-08");
        user1.setSchool("北大");

        User user2 = new User();
        user2.setName("王五");
        user2.setAge(25);
        user2.setBirthday("1995-02-05");
        user2.setSchool("北大");

        User user3 = new User();
        user3.setName("赵六");
        user3.setAge(43);
        user3.setBirthday("1977-04-03");
        user3.setSchool("复旦");

        User user4 = new User();
        user4.setName("张三丰");
        user4.setAge(80);
        user4.setBirthday("1940-08-15");
        user4.setSchool("复旦");

        User user5 = new User();
        user5.setName("王重阳");
        user5.setAge(70);
        user5.setBirthday("1950-07-07");
        user5.setSchool("清华");

        Object[] users = new Object[]{user1, user2, user3, user4, user5};

        BulkRequest bulkRequest = new BulkRequest();
        bulkRequest.timeout("5s");

        for (int i = 0; i < users.length; i++) {
            String id = String.valueOf(i + 2);
            String source = JSON.toJSONString(users[i]);
            bulkRequest.add(new IndexRequest("user").id(id).source(source, XContentType.JSON));
        }

        BulkResponse responses = client.bulk(bulkRequest, RequestOptions.DEFAULT);
        System.out.println(responses.status());
    }

接下来通过单元测试创建索引、添加文档：

@RunWith(SpringRunner.class)
@SpringBootTest
class LearnElasticsearchApplicationTests {

    @Autowired
    UserService userService;

    @Test
    void testES() throws IOException {
        userService.createIndex2();
        userService.addDocument();
        userService.bulkAddDocument();
    }
}

最终在 head 工具中可以看到如下数据：

5、修改文档

可以根据文档 id 修改文档，修改前可以判断文档是否存在：

public boolean existsDocument() throws IOException {
    GetRequest request = new GetRequest("user", "1");
    // 不获取_source的内容
    request.fetchSourceContext(new FetchSourceContext(false));
    // 不获取已排序字段
    request.storedFields("_none_");
    boolean exists = client.exists(request, RequestOptions.DEFAULT);
    return exists;
}

修改 id 为 1 的文档的age字段值：

public void updateDocument() throws IOException {
    if (!existsDocument()) {
        return;
    }
    UpdateRequest request = new UpdateRequest("user", "1");
    User user = new User();
    user.setAge(31);
    updateRequest.doc(JSON.toJSONString(user), XContentType.JSON);
    UpdateResponse response = client.update(request, RequestOptions.DEFAULT);
    System.out.println(response.status());
}

根据文档 id 修改的局限性还是太强了，我们还可以使用UpdateByQueryRequest根据查询条件来批量修改文档：

public void updateDocument2() throws IOException {
    UpdateByQueryRequest request = new UpdateByQueryRequest("user");
    // 设置查询条件
    request.setQuery(new MatchPhraseQueryBuilder("name", "张三"));
    // request.setQuery(new TermQueryBuilder("name.keyword", "张三"));
    // 设置一次可以批处理的文档数，默认1000
    request.setBatchSize(200);
    // 更新后刷新索引
    request.setRefresh(true);
    // 通过脚本设置如何更新
    request.setScript(new Script("ctx._source.school = '复旦'"));
    BulkByScrollResponse response = client.updateByQuery(request, RequestOptions.DEFAULT);
    System.out.println("修改的文档数：" + response.getStatus().getUpdated());
}

6、删除文档

首先可以根据文档 id 来删除文档：

public void deleteDocument() throws IOException {
    if (!existsDocument()) {
        return;
    }
    DeleteRequest request = new DeleteRequest("user", "1");
    DeleteResponse response = client.delete(request, RequestOptions.DEFAULT);
    System.out.println(response.status());
}

和修改类似，我们也可以使用DeleteByQueryRequest批量删除符合指定查询条件的文档：

public void deleteDocument2() throws IOException {
    DeleteByQueryRequest request = new DeleteByQueryRequest("user");
    // 设置查询条件，查询school是复旦的
    request.setQuery(new TermQueryBuilder("school.keyword", "复旦"));
    // request.setQuery(new MatchPhraseQueryBuilder("school", "复旦"));
    // 设置一次可以批处理的文档数，默认1000
    request.setBatchSize(200);
    // 更新后刷新索引
    request.setRefresh(true);
    BulkByScrollResponse response = client.deleteByQuery(request, RequestOptions.DEFAULT);
    System.out.println("删除的文档数：" + response.getStatus().getDeleted());
}