本文由 简悦 SimpRead 转码， 原文地址 blog.csdn.net
目录
概要
ES 通过工具类
最近项目中由于数据量大，考虑上 ES，但是找了很多资料发现没有比较理想的内容，最终决定自己写一个，话不多说，直接上代码
ES 安装
下载地址：Past Releases of Elastic Stack Software | Elastic
选择 8.14.1 版本下载
因为这个版本的 es 是自带安全认证的，所以如果是需要 http 访问的话，需要自己在 config 下修改配置文件
打开 elasticsearch.yml 将安全认证注释掉，然后改为 false 就可以了
xpack.security.enabled: false
xpack.security.enrollment.enabled: false
xpack.security.http.ssl:
enabled: false
xpack.security.transport.ssl:
enabled: false
ES 连接工具 kibana
下载地址：Download Kibana Free | Get Started Now | Elastic
下载之后在 config 目录下打开配置文件 kibana.yml，将 elasticsearch.hosts 改成 es 对应的地址即可，此处不做详细说明
源码
一、建立与 es 的连接
@Configuration
public class ElasticSearchConfig {
@Value("${spring.elasticsearch.uris}")
private String hosts;
@Value("${spring.elasticsearch.username}")
private String userName;
@Value("${spring.elasticsearch.password}")
private String passWord;
@Bean()
public ElasticsearchClient elasticsearchClient(){
HttpHost[] httpHosts = toHttpHost();
final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
credentialsProvider.setCredentials(
AuthScope.ANY, new UsernamePasswordCredentials(userName, passWord));
RestClientBuilder builder = RestClient.builder(httpHosts);
builder.setRequestConfigCallback(
requestConfigBuilder -> requestConfigBuilder.setSocketTimeout(60000).setConnectTimeout(5000));
builder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
@Override
public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) {
return httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
}
});
RestClient restClient = builder.build();
ElasticsearchTransport transport = new RestClientTransport(restClient,new JacksonJsonpMapper());
return new ElasticsearchClient(transport);
}
private HttpHost[] toHttpHost() {
if (!StringUtils.hasLength(hosts)) {
throw new RuntimeException("invalid elasticsearch configuration. elasticsearch.hosts不能为空！");
}
// 多个IP逗号隔开
String[] hostArray = hosts.split(",");
HttpHost[] httpHosts = new HttpHost[hostArray.length];
HttpHost httpHost;
for (int i = 0; i < hostArray.length; i++) {
String[] strings = hostArray[i].split(":");
httpHost = new HttpHost(strings[0], Integer.parseInt(strings[1]), "http");
httpHosts[i] = httpHost;
}
return httpHosts;
}
}
二、下边实现了 es 新增删除索引、以及单个插入，批量插入，和通过脚本查询接口
@Slf4j
@Service
public class IElasticsearchService {
@Resource
private ElasticsearchClient client;
/**
* 判断索引是否存在.
*
* @param indexName index名称
*/
public boolean existIndex(String indexName) {
try {
BooleanResponse booleanResponse = client.indices().exists(e -> e.index(indexName));
return !booleanResponse.value();
} catch (IOException e) {
log.error("向es中检测索引【{}】出错，错误信息为：{}", indexName, e.getMessage());
}
return true;
}
/**
* 创建索引.
*
* @param indexName index名称
*/
public void createIndex(String indexName) {
try {
client.indices().create(c -> c.index(indexName));
} catch (IOException e) {
log.error("向es中创建索引【{}】出错，错误信息为：{}", indexName, e.getMessage());
}
}
/**
* 删除索引.
*
* @param indexName index名称
*/
public void deleteIndex(String indexName) {
try {
client.indices().delete(c -> c.index(indexName));
} catch (IOException e) {
log.error("向es中删除索引【{}】出错，错误信息为：{}", indexName, e.getMessage());
}
}
/**
* 添加记录.
*
*/
public <T> void addDocument(T param, String indexName) {
try {
if (this.existIndex(indexName)) {
this.createIndex(indexName);
}
client.index(i -> i.index(indexName).id(getIdFromItem(param)).document(param));
} catch (IOException e) {
log.error("向es中添加Document出错!{}", e.getMessage());
}
}
/**
* 批量添加.
*
* @param hisList 添加的数量集合
* @param indexName 索引名称
*/
public <T> void batchAddDocument(List<T> hisList, String indexName) {
if (this.existIndex(indexName)) {
this.createIndex(indexName);
}
BulkRequest.Builder br = new BulkRequest.Builder();
hisList.forEach(t -> br.operations(op -> op
.index(idx -> idx
.index(indexName)
.id(getIdFromItem(t))
.document(t)
))
);
try {
BulkResponse result = client.bulk(br.build());
if (result.errors()) {
log.error("Bulk had errors");
for (BulkResponseItem item : result.items()) {
if (item.error() != null) {
log.error(item.error().reason());
}
}
}
} catch (IOException e) {
log.error("向es中添加Document出错,{}", e.getMessage());
}
}
/**
* 根据索引名称和字段查询数据.
*
* @param indexName 索引名称
*/
public <T> EsPair<T> findDocumentByField(String indexName, String script, Class<T> clazz) {
try {
client.putScript(r -> r
.id("query-script")
.script(s -> s
.lang("mustache")
.source(script)
));
SearchTemplateResponse<T> response = client.searchTemplate(r -> r
.index(indexName)
.id("query-script"),
clazz
);
List<Hit<T>> hitList = response.hits().hits();
long count = 0;
if (response.hits().total() != null) {
count = response.hits().total().value();
}
List<T> hisList = new ArrayList<>();
for (Hit<T> mapHit : hitList) {
hisList.add(mapHit.source());
}
return new EsPair<>(hisList, count);
} catch (IOException e) {
log.error("【查询 -> 失败】从es中查询分析后的日志出错，错误信息为：{}", e.getMessage());
}
return null;
}
/**
* 通过id批量删除
* @param indexName 索引名称
* @param ids id集合
*/
public void deleteDocumentById(String indexName, List<String> ids) {
List<FieldValue> values = new ArrayList<>();
ids.forEach(h -> values.add(FieldValue.of(h)));
Query idsQuery = TermsQuery.of(t -> t.field("id").terms(new TermsQueryField.Builder()
.value(values).build()
))._toQuery();
try {
client.deleteByQuery(t -> t
.index(indexName)
.query(idsQuery));
} catch (IOException e) {
throw new RuntimeException(e);
}
}
// 这个方法假设你的对象有getId()方法，实际中请根据对象结构进行调整
private <T> String getIdFromItem(T item) {
if (item instanceof Map) {
// 如果item是一个Map，直接尝试获取"id"键的值
Map<?, ?> map = (Map<?, ?>) item;
return (String) map.get("id");
} else {
// 对于非Map对象，继续使用反射尝试调用getId方法
try {
Method getIdMethod = item.getClass().getMethod("getId");
return (String) getIdMethod.invoke(item);
} catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
throw new IllegalArgumentException("对象缺少getId方法或者执行时发生错误", e);
}
}
}
}
三、当我做到这儿的时候面临的问题也就来了，脚本怎么能做成通用的动态的呢，然后我就写了个工具类。下边这个工具类就可以动态的添加条件了，额 ···· 就和动态拼接 SQL 的道理是一样的
@Resource
private IElasticsearchService elasticsearchService;
// 创建EsQuery
EsQueryUtil esQueryUtil = new EsQueryUtil(startLine, pageSize);
// code集合
Set<String> codeSet = new HashSet<>();
if (!codeSet.isEmpty()) {
// 添加code精确查询
esQueryUtil.addTerms("code", codeSet);
}
// 添加name模糊查询
esQueryUtil.addMatch("name", user.getName());
// 添加时间范围
esQueryUtil.addRange("add_time", start_time == 0 ? null : start_time, end_time == 0 ? null : end_time);
// 添加排序
esQueryUtil.addSort("add_time","desc");
esQueryUtil.addSort("sort","desc");
// 深分页时需传入最后一条数据的排序内容
if (StringUtil.isNotBlank(user.getAdd_time()) && StringUtil.isNotBlank(user.getSort())) {
esQueryUtil.addAfter(user.getAdd_time());
esQueryUtil.addAfter(user.getSort());
}
EsPair<Map> esPair = elasticsearchService.findDocumentByField(indexName, esQueryUtil.getScript(false), Map.class);
public class EsQueryUtil {
private Map<String, Object> bool;
private List<Map<String, Object>> must;
private List<Map<String, Object>> sorts;
private List<Map<String, Object>> ranges;
private Integer from;
private Integer size;
private List<Object> after;
public EsQueryUtil() {
init();
}
public EsQueryUtil(int from, int size) {
this.from = from;
this.size = size;
init();
}
private void init() {
bool = new HashMap<>();
must = new ArrayList<>();
sorts = new ArrayList<>();
ranges = new ArrayList<>();
after = new ArrayList<>();
bool.put("must", must);
}
/**
* 添加精确查询条件
* @param field 字段
* @param values 值 set可防止条件重复
*/
public void addTerms(String field, Set<String> values) {
Map<String, Object> termClause = new HashMap<>();
termClause.put("terms", new HashMap<String, Object>() {{
put(field + ".keyword", values);
}});
must.add(termClause);
}
/**
* 添加模糊查询条件
* @param field 字段
* @param value 值
*/
public void addMatch(String field, String value) {
Map<String, Object> matchClause = new HashMap<>();
matchClause.put("match", new HashMap<String, Object>() {{
put(field + ".keyword", value);
}});
must.add(matchClause);
}
/**
* 添加排序条件
* @param field 字段
* @param order 值
*/
public void addSort(String field, String order) {
Map<String, Object> sortClause = new HashMap<>();
sortClause.put(field, order.equals("asc") ? "asc" : "desc");
sorts.add(sortClause);
}
/**
* 添加范围条件
* @param field 值
* @param gt 开始
* @param lt 结束
*/
public void addRange(String field, Object gt, Object lt) {
Map<String, Object> rangeClause = new HashMap<>();
rangeClause.put("range", new HashMap<String, Object>() {{
put(field, new HashMap<String, Object>() {{
if (gt != null) {
put("gt", gt);
}
if (lt != null) {
put("lt", lt);
}
}});
}});
// 根据需求决定将范围查询添加到must、filter或其他bool子句中
// 以下示例是添加到must中，根据实际情况调整
ranges.add(rangeClause);
}
/**
* 添加最后一条数据值，与排序字段顺序对应
* @param data 值
*/
public void addAfter(Object data) {
after.add(data);
}
/**
* 提取脚本
* @param isCount 是否查询总数
* @return 结果
*/
public String getScript(boolean isCount) {
// 确保terms和match条件位于bool的must子句中
bool.put("must", must);
// 将范围查询条件放入filter子句，如果存在的话
if (!ranges.isEmpty()) {
bool.put("filter", ranges);
}
Map<String, Object> query = new HashMap<>();
query.put("bool", bool);
Map<String, Object> finalQuery = new HashMap<>();
if (!sorts.isEmpty()) {
finalQuery.put("sort", sorts);
}
if (!isCount) {
// 大于1w判断，如果大于1w，需要使用search_after(游标)进行分页
if (from + size > 10000) {
finalQuery.put("search_after", after);
} else {
finalQuery.put("from", from);
}
finalQuery.put("size", size);
}
finalQuery.put("query", query);
finalQuery.put("track_total_hits", true);
try {
return new ObjectMapper().writeValueAsString(finalQuery);
} catch (JsonProcessingException e) {
return null;
}
}
}
四、 这个是返回值的类，因为我不想再查询一遍总数了，所以又写了个返回值的类
@Getter
public class EsPair<T> {
private final List<T> list;
private final long count;
public EsPair(List<T> list, long count) {
this.list = list;
this.count = count;
}
}
小结
如此一来，es 就变成通用的了，可以传入任何索引和实体，查询分页或者非分页的数据，但是 ES 的版本之间变化有点大，多少版本能适用就不大清楚了，但是思路应该都是差不多的
如果查询方法直接对外开放的话还可以充当一部分 kibana 的功能
@ApiOperation(value = "执行ES脚本")
@PostMapping(value = "runEsScript")
public AjaxResult runEsScript(
@ApiParam(value = "索引名称", name = "indexName", required = true)
@RequestParam String indexName,
@ApiParam(value = "脚本", name = "script", required = true)
@RequestParam String script
) {
EsPair<Map> esPair = elasticsearchService.findHisByField(indexName, script, Map.class);
AjaxResult ajaxResult = AjaxResult.success();
ajaxResult.put("count", esPair.getCount());
ajaxResult.put("data", esPair.getList());
return ajaxResult;
}
对了，中途还碰到个小问题，就是由于 jackson 包的版本冲突问题
ES-8.14.1 需要引入 2.17.0 的版本，但是我项目中有别的地方引入了 2.11.4 的版本，所以最后只好手动给他过滤掉了
还有就是这种 from，size 分页只能到 1w 条，超过 1w 的深分页就要用别的方式了，我在工具类中使用的是 search_after 的方式，需要传入当前最后一条数据的排序字段，只不过后边就不能跳转了，如果必须要跳转的话建议在点击下一页时缓存条件、页数与每一页最后一条数据的排序值，这样也可以实现后续的跳转
<exclusions>
<exclusion>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</exclusion>
<exclusion>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-annotations</artifactId>
</exclusion>
<exclusion>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-core</artifactId>
</exclusion>
</exclusions>