使用JestClient操作ElasticSearch
胥良平
1. 前言
ElasticSearch是一个在全文搜索引擎库Apache Lucene基础之上建立的开源服务，它提供了一个分布式、高扩展、高实时的搜索与数据分析引擎。
在Spring Boot中集成ElasticSearch有Spring Data Elasticsearch、REST Client和Jest等方式。其中Jest作为一个用于ElasticSearch的HTTP Java 客户端，提供了更流畅的API和更容易使用的接口。
本文将介绍Jest的基本用法。
2. Maven依赖
Maven依赖
<dependency>
<groupId>io.searchbox</groupId>
<artifactId>jest</artifactId>
<version>5.3.3</version>
</dependency>
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
<version>6.2.4</version>
</dependency>
3. 配置文件
配置比较简单，只需要在application.yml中添加相应配置：
spring:
elasticsearch:
jest:
uris: https://ceshi.elastic.feibo.com
username: elastic
password: 123456
read-timeout: 1000s
connection-timeout: 1000s
multi-threaded: true
4. 操作
由于添加了Jest配置，只需要注入JestClient就可以直接使用，该客户端连接到本地运行的Elasticsearch。
JestClient类是通用类，只有少数公共方法。我们将使用的一个主要方法是execute，它接受Action接口的一个实例。Jest客户端提供了几个构建器类来帮助创建与ElasticSearch交互的不同操作：
@Resource
private JestClient jestClient;
所有Jest调用的结果都是JestResult的一个实例。我们可以定义一个checkResult方法来检查是否成功。对于失败的操作，我们可以调用getJsonString方法来获取更多详细信息或进行其他操作：
private boolean checkRes(JestResult jestResult) {
if (!jestResult.isSucceeded()) {
logger.error(jestResult.getResponseCode() + jestResult.getJsonString());
throw new RuntimeException(jestResult.getJsonString());
}
return true;
}
4.1 索引管理
创建索引使用CreateIndex操作，可以选择是否手动设置settings值：
public boolean createIndex(String indexName, String settings) {
try {
JestResult jestResult;
if (null == settings) {
jestResult = jestClient.execute(new CreateIndex.Builder(indexName).build());
} else {
jestResult = jestClient.execute(new CreateIndex.Builder(indexName).settings(Settings.builder().loadFromSource(settings, XContentType.JSON).build()).build());
}
return checkRes(jestResult);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
设置mapping使用PutMapping操作：
public boolean createIndexMapping(String indexName, String typeName, String mappingString) {
try {
JestResult jestResult = jestClient.execute(new PutMapping.Builder(indexName, typeName, mappingString).build());
return checkRes(jestResult);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
4.2 文档管理
Jest可以使用最基本的Json字符串格式进行储存，也可以接受表示要索引的文档的任何POJO。为了方便文档操作，我们可以先创建一个DTO类来定义文档：
package com.xx.xx.bean.dto.common;
import io.searchbox.annotations.JestId;
import io.searchbox.annotations.JestVersion;
/**
* ES文档对应字段
*
* @author LKET
* @date 2019/6/21 下午2:50
*/
public class StatisticsDTO {
/**
* 文档id
*/
@JestId
private String documentId;
/**
* 文档版本
*/
@JestVersion
private Long documentVersion;
/**
* 页面名称
*/
private String pageName;
/**
* 按钮位置名称
*/
private String locationName;
/**
* 用户id
*/
private Integer userId;
/**
* 数量
*/
private Integer quantity;
/**
* 创建时间
*/
private Long createdAt;
public String getDocumentId() {
return documentId;
}
public void setDocumentId(String documentId) {
this.documentId = documentId;
}
public Long getDocumentVersion() {
return documentVersion;
}
public void setDocumentVersion(Long documentVersion) {
this.documentVersion = documentVersion;
}
public String getPageName() {
return pageName;
}
public void setPageName(String pageName) {
this.pageName = pageName;
}
public String getLocationName() {
return locationName;
}
public void setLocationName(String locationName) {
this.locationName = locationName;
}
public Integer getUserId() {
return userId;
}
public void setUserId(Integer userId) {
this.userId = userId;
}
public Integer getQuantity() {
return quantity;
}
public void setQuantity(Integer quantity) {
this.quantity = quantity;
}
public Long getCreatedAt() {
return createdAt;
}
public void setCreatedAt(Long createdAt) {
this.createdAt = createdAt;
}
}
创建、更新、删除文档分别使用Index、Update、Delete操作：
public boolean insert(String indexName, String typeName, StatisticsDTO source) {
try {
JestResult jestResult = jestClient.execute(new Index.Builder(source).index(indexName).type(typeName).build());
return checkRes(jestResult);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
public boolean update(String indexName, String typeName, String id) {
try {
JestResult jestResult = jestClient.execute(new Update.Builder(id).index(indexName).type(typeName).build());
return checkRes(jestResult);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
public boolean delete(String indexName, String typeName, String id) {
try {
JestResult jestResult = jestClient.execute(new Delete.Builder(id).index(indexName).type(typeName).build());
return checkRes(jestResult);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
查询文档分为两种方式，一种是知道文档的id，根据id查询文档使用Get操作，在返回搜索结果时可以直接获取原始的JSON格式，也可以用getSourceAsObject方法来转为DTO；条件搜索查询使用Search操作，同样的在获取查询结果时用getSourceAsObjectList方法来转为数组DTO：
public StatisticsDTO getById(String indexName, String typeName, String id) {
try {
JestResult jestResult = jestClient.execute(new Get.Builder(indexName, id).type(typeName).build());
if (jestResult != null && jestResult.isSucceeded()) {
return jestResult.getSourceAsObject(StatisticsDTO.class);
}
} catch (Exception e) {
throw new RuntimeException(e);
}
return null;
}
public List<StatisticsDTO> search(String indexName, String typeName, String query) {
List<StatisticsDTO> statisticsDTOList = new ArrayList<>();
try {
JestResult jestResult = jestClient.execute(new Search.Builder(query).addIndex(indexName).addType(typeName).build());
if (jestResult != null && jestResult.isSucceeded()) {
return jestResult.getSourceAsObjectList(StatisticsDTO.class);
}
} catch (Exception e) {
throw new RuntimeException(e);
}
return statisticsDTOList;
}
4.3 批量操作
Jest同样支持批量操作，甚至可以用不同类型的请求组合在一起，将任意数量的请求组合成单个调用，同时发送多个操作来节省时间：
JestResult jestResult = jestClient.execute(new Bulk.Builder()
.defaultIndex(indexName)
.defaultType(typeName)
.addAction(new Index.Builder(insertSource).build())
.addAction(new Update.Builder(updateSource).id(updateId).build())
.addAction(new Delete.Builder(deleteId).build())
.build());
4.4 异步操作
Jest还支持异步操作。这意味着我们可以使用非阻塞I/O执行上述任何操作。要异步调用操作，只需使用客户端的executeAsync方法：
JestResult jestResult = jestclient.executeAsync(
new Index.Builder(insertSource).build(),
new JestResultHandler<JestResult>() {
@Override
public void completed(JestResult result) {
// handle result
}
@Override
public void failed(Exception ex) {
// handle exception
}
});
5. 完整代码及测试类
组件：
package com.xx.xx.component;
import com.xx.xx.bean.dto.common.StatisticsDTO;
import io.searchbox.client.JestClient;
import io.searchbox.client.JestResult;
import io.searchbox.core.*;
import io.searchbox.indices.CreateIndex;
import io.searchbox.indices.mapping.GetMapping;
import io.searchbox.indices.mapping.PutMapping;
import io.searchbox.indices.settings.GetSettings;
import io.searchbox.indices.settings.UpdateSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
/**
* Jest组件
*
* @author LKET
* @date 2019/5/30 下午5:53
*/
@Component
public class JestClientComponent {
private final Logger logger = LoggerFactory.getLogger(JestClientComponent.class);
@Resource
private JestClient jestClient;
/**
* 检验是否执行错误
*
* @param jestResult jestResult
* @author LKET
* @date 2019/6/21 下午4:46
*/
private boolean checkRes(JestResult jestResult) {
if (!jestResult.isSucceeded()) {
logger.error(jestResult.getResponseCode() + jestResult.getJsonString());
throw new RuntimeException(jestResult.getJsonString());
}
return true;
}
/**
* 创建索引
*
* @param indexName 索引名称
* @param settings json格式的设置(传null为默认设置；传值示例：{"number_of_shards":4,"number_of_replicas":1})
* @author LKET
* @date 2019/6/21 下午4:29
*/
public boolean createIndex(String indexName, String settings) {
try {
JestResult jestResult;
if (null == settings) {
jestResult = jestClient.execute(new CreateIndex.Builder(indexName).build());
} else {
jestResult = jestClient.execute(new CreateIndex.Builder(indexName).settings(Settings.builder().loadFromSource(settings, XContentType.JSON).build()).build());
}
return checkRes(jestResult);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
/**
* 删除索引
*
* @param indexName 索引名称
* @author LKET
* @date 2019/6/21 下午4:29
*/
public boolean deleteIndex(String indexName) {
try {
JestResult jestResult = jestClient.execute(new Delete.Builder(indexName).build());
return checkRes(jestResult);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
/**
* 获取索引的setting
*
* @param indexName 索引名称
* @author LKET
* @date 2019/6/24 下午4:58
*/
public String getIndexSettings(String indexName) {
try {
JestResult jestResult = jestClient.execute(new GetSettings.Builder().addIndex(indexName).build());
if (jestResult != null && jestResult.isSucceeded()) {
return jestResult.getJsonString();
}
} catch (Exception e) {
throw new RuntimeException(e);
}
return null;
}
/**
* 修改索引的setting
*
* @param indexName 索引名称
* @param settings json格式的设置(传值示例：{"max_result_window":"10000"})
* @author LKET
* @date 2019/6/25 下午4:00
*/
public boolean updateIndexSettings(String indexName, String settings) {
try {
JestResult jestResult = jestClient.execute(new UpdateSettings.Builder(settings).addIndex(indexName).build());
return checkRes(jestResult);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
/**
* 设置索引的mapping
*
* @param indexName 索引名
* @param typeName 类型名
* @param mappingString json格式的mapping串(传值示例：{"typeName":{"properties":{"message":{"type":"string","store":"yes"}}}})
* @author LKET
* @date 2019/6/21 下午4:48
*/
public boolean createIndexMapping(String indexName, String typeName, String mappingString) {
try {
JestResult jestResult = jestClient.execute(new PutMapping.Builder(indexName, typeName, mappingString).build());
return checkRes(jestResult);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
/**
* 获取索引的mapping
*
* @param indexName 索引名
* @param typeName 类型名
* @author LKET
* @date 2019/6/24 下午4:45
*/
public String getIndexMapping(String indexName, String typeName) {
try {
JestResult jestResult = jestClient.execute(new GetMapping.Builder().addIndex(indexName).addType(typeName).build());
if (jestResult != null && jestResult.isSucceeded()) {
return jestResult.getJsonString();
}
} catch (Exception e) {
throw new RuntimeException(e);
}
return null;
}
/**
* 添加文档
*
* @param indexName 索引名
* @param typeName 类型名
* @param source 文档内容
* @author LKET
* @date 2019/6/21 下午3:02
*/
public boolean insert(String indexName, String typeName, StatisticsDTO source) {
try {
JestResult jestResult = jestClient.execute(new Index.Builder(source).index(indexName).type(typeName).build());
return checkRes(jestResult);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
/**
* 更新文档
*
* @param indexName 索引名
* @param typeName 类型名
* @param id 文档id
* @author LKET
* @date 2019/6/21 下午3:30
*/
public boolean update(String indexName, String typeName, String id) {
try {
JestResult jestResult = jestClient.execute(new Update.Builder(id).index(indexName).type(typeName).build());
return checkRes(jestResult);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
/**
* 删除文档
*
* @param indexName 索引名
* @param typeName 类型名
* @param id 文档id
* @author LKET
* @date 2019/6/21 下午3:30
*/
public boolean delete(String indexName, String typeName, String id) {
try {
JestResult jestResult = jestClient.execute(new Delete.Builder(id).index(indexName).type(typeName).build());
return checkRes(jestResult);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
/**
* 根据id获取文档
*
* @param indexName 索引名
* @param typeName 类型名
* @param id 文档id
* @author LKET
* @date 2019/6/25 下午1:30
*/
public StatisticsDTO getById(String indexName, String typeName, String id) {
try {
JestResult jestResult = jestClient.execute(new Get.Builder(indexName, id).type(typeName).build());
if (jestResult != null && jestResult.isSucceeded()) {
return jestResult.getSourceAsObject(StatisticsDTO.class);
}
} catch (Exception e) {
throw new RuntimeException(e);
}
return null;
}
/**
* 条件查询
*
* @param indexName 索引名
* @param typeName 类型名
* @param query 查询语句(传值示例：{"query":{"bool":{"must":[{"match":{"pageName":"homePage"}},{"match":{"locationName":"skip"}}]}}})
* @author LKET
* @date 2019/6/25 下午1:30
*/
public List<StatisticsDTO> search(String indexName, String typeName, String query) {
List<StatisticsDTO> statisticsDTOList = new ArrayList<>();
try {
JestResult jestResult = jestClient.execute(new Search.Builder(query).addIndex(indexName).addType(typeName).build());
if (jestResult != null && jestResult.isSucceeded()) {
return jestResult.getSourceAsObjectList(StatisticsDTO.class);
}
} catch (Exception e) {
throw new RuntimeException(e);
}
return statisticsDTOList;
}
/**
* 聚合搜索查询
*
* @param indexName 索引名
* @param typeName 类型名
* @param query 查询语句(传值示例：{"query":{"bool":{"must":[{"match":{"pageName":"homePage"}},{"match":{"locationName":"skip"}}]}},"aggs":{"distinct":{"cardinality":{"field":"userId"}}}})
* @author LKET
* @date 2019/6/26 上午11:25
*/
public SearchResult searchAggregations(String indexName, String typeName, String query) {
try {
SearchResult searchResult = jestClient.execute(new Search.Builder(query).addIndex(indexName).addType(typeName).build());
if (searchResult != null && searchResult.isSucceeded()) {
return searchResult;
}
} catch (Exception e) {
throw new RuntimeException(e);
}
return null;
}
}
测试类：
package xx.fb.explosive;
import com.xx.xx.bean.dto.common.StatisticsDTO;
import com.xx.xx.component.JestClientComponent;
import io.searchbox.core.SearchResult;
import io.searchbox.core.search.aggregation.*;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import javax.annotation.Resource;
import java.util.List;
/**
* Jest单元测试
*
* @author LKET
* @date 2019/6/24 下午4:33
*/
@RunWith(SpringRunner.class)
@SpringBootTest
public class JestComponentTest {
@Resource
private JestClientComponent jestClientComponent;
private static final String INDEX_NAME = "test_index";
private static final String TYPE_NAME = "test_type";
@Test
public void testCreateIndex() {
boolean res = jestClientComponent.createIndex(INDEX_NAME, null);
Assert.assertTrue(res);
}
@Test
public void testCreateIndexWithSettings() {
boolean res = jestClientComponent.createIndex(INDEX_NAME, "{\"number_of_shards\":5,\"number_of_replicas\":1}");
Assert.assertTrue(res);
}
@Test
public void testGetIndexSettings() {
String res = jestClientComponent.getIndexSettings(INDEX_NAME);
Assert.assertNotNull(res);
}
@Test
public void testUpdateIndexSettings() {
boolean res = jestClientComponent.updateIndexSettings(INDEX_NAME, "{\"max_result_window\":\"10000\"}");
Assert.assertTrue(res);
}
@Test
public void testDeleteIndex() {
boolean res = jestClientComponent.deleteIndex(INDEX_NAME);
Assert.assertTrue(res);
}
@Test
public void testCreateIndexMapping() {
String mappingString = "{\"" + TYPE_NAME + "\":{\"properties\":{\"pageName\":{\"type\":\"text\"},\"locationName\":{\"type\":\"text\"},\"userId\":{\"type\":\"integer\"},\"quality\":{\"type\":\"integer\"},\"createdAt\":{\"type\":\"long\"}}}}";
boolean res = jestClientComponent.createIndexMapping(INDEX_NAME, TYPE_NAME, mappingString);
Assert.assertTrue(res);
}
@Test
public void testGetMapping() {
String res = jestClientComponent.getIndexMapping(INDEX_NAME, TYPE_NAME);
Assert.assertNotNull(res);
}
@Test
public void testInsert() {
StatisticsDTO source = new StatisticsDTO();
source.setPageName("homePage");
source.setLocationName("skip");
source.setUserId(2);
source.setQuantity(1);
source.setCreatedAt(System.currentTimeMillis() / 1000);
boolean res = jestClientComponent.insert(INDEX_NAME, TYPE_NAME, source);
Assert.assertTrue(res);
}
@Test
public void testGetById() {
StatisticsDTO res = jestClientComponent.getById(INDEX_NAME, TYPE_NAME, "NwuWh2sBdr3hviwhgnr9");
Assert.assertNotNull(res);
}
@Test
public void testSearch() {
List<StatisticsDTO> res = jestClientComponent.search(INDEX_NAME, TYPE_NAME, "{\"query\":{\"bool\":{\"must\":{\"match\":{\"userId\":\"1\"}}}}}");
Assert.assertNotNull(res);
}
@Test
public void testSearchAggregations() {
// 去重查询（类似distinct）
SearchResult distinctResult = jestClientComponent.searchAggregations(INDEX_NAME, TYPE_NAME, "{\"query\":{\"bool\":{\"must\":[{\"match\":{\"pageName\":\"homePage\"}},{\"match\":{\"locationName\":\"skip\"}}]}},\"aggs\":{\"distinct\":{\"cardinality\":{\"field\":\"userId\"}}}}");
// 获取distinct后的数值
CardinalityAggregation cardinalityAggregation = distinctResult.getAggregations().getCardinalityAggregation("distinct");
Long distinctValue = cardinalityAggregation.getCardinality();
Assert.assertNotNull(distinctValue);
}
}
6. 总结
在本篇文章中，简要介绍了Jest中一小部分功能，但很明显Jest是一个健壮的Elasticsearch客户端。它的流畅的构建器类和RESTful接口使其易于学习，并且它对Elasticsearch接口的完全支持使其成为原生客户端的一个有力的替代方案。
类似资料：