SpringBoot整合Elasticsearch游标查询的示例代码(scroll)

网友投稿 388 2022-11-17


SpringBoot整合Elasticsearch游标查询的示例代码(scroll)

游标查询(scroll)简介

scroll 查询 可以用来对 Elasticsearch 有效地执行大批量的文档查询,而又不用付出深度分页那种代价。

游标查询会取某个时间点的快照数据。 查询初始化之后索引上的任何变化会被它忽略。 它通过保存旧的数据文件来实现这个特性,结果就像保留初始化时的索引 视图 一样。

启用游标查询可以通过在查询的时候设置参数 scroll 的值为我们期望的游标查询的过期时间。 游标查询的过期时间会在每次做查询的时候刷新,所以这个时间只需要足够处理当前批的结果就可以了,而不是处理查询结果的所有文档的所需时间。 这个过期时间的参数很重要,因为保持这个游标查询窗口需要消耗资源,所以我们期望如果不再需要维护这种资源就该早点儿释放掉。 设置这个超时能够让 Elasticsearch 在稍后空闲的时候自动释放这部分资源。

GET /old_index/_search?scroll=1m

{

"query": { "match_all": {}},

"sort" : ["_doc"],

"size": 1000

}

scroll=1m:保持游标查询窗口一分钟。

返回结果示例:

{

"_scroll_id": "cXVlcnlUaGVuRmV0Y2g7NTsxMDk5NDpkUmpiR2FjOFNhNnlCM1ZDMWpWYnRROzEwOTk1OmRSamJHYWM4U2E2eUIzVkMxalZidFE7MTA5OTM6ZFJqYkdhYzhTYTZ5QjNWQzFqVmJ0UTsxMTE5MDpBVUtwN2lxc1FLZV8yRGVjWlI2QUVBOzEwOTk2OmRSamJHYWM4U2E2eUIzVkMxalZidFE7MDs=",

"took": 10,

"timed_out": false,

"_shards": {

"total": 5,

"successful": 5,

"failed": 0

},

"hits": {

"total": 2633253,

"max_score": 1.0,

"hits": [

{

"_index": "old_index",

"_type": "old_index_type",

"_id": "1",

"_score": 1.0,

"_source": {

...

}

}

]

}

}

这个查询的返回结果包括一个字段 _scroll_id, 它是一个base64编码的长字符串 。 现在我们能传递字段 _scroll_id 到 _search/scroll 查询接口获取下一批结果:

GET /_search/scroll

{

"scroll": "1m",

"scroll_id" : "cXVlcnlUaGVuRmV0Y2g7NTsxMDk5NDpkUmpiR2FjOFNhNnlCM1ZDMWpWYnRROzEwOTk1OmRSamJHYWM4U2E2eUIzVkMxalZidFE7MTA5OTM6ZFJqYkdhYzhTYTZ5QjNWQzFqVmJ0UTsxMTE5MDpBVLUeedgazUtwN2lxc1FLZV8yRGVjWlI2QUVBOzEwOTk2OmRSamJHYWM4U2E2eUIzVkMxalZidFE7MDs="

}

注意:需要再次设置游标查询过期时间为一分钟。

这个游标查询返回下一批结果。

另外尽管我们指定字段 size 的值为1000,但是我们有可能取到超过这个值数量的文档。 当查询的时候, 字段 size 作用于单个分片,所以每个批次实际返回的文档数量最大为 size * number_of_primary_shards。

注意:游标查询每次返回一个新字段 _scroll_id。每次我们做下一次游标查询, 我们必须把前一次查询返回的字段_scroll_id 传递进去。 当没有更多的结果返回的时候,我们就处理完所有匹配的文档了。

整合

新增以下三个方法:

/**

* 游标查询

* @param params 查询入参

* @param indexName 索引名称

* @param type 索引类型

* @param defaultSort 默认排序

* @param keyMappings 字段映射

* @param keyMappingsMap 索引对应字段映射

* @param scrollTimeInMillis 游标开启的时间

* @return Page

*/

protected Page commonStartScroll(Map params, String indexName, String type, String defaultSort,

Map keyMappings,

Map> keyMappingsMap, long scrollTimeInMillis) {

SearchQuery searchQuery = buildSearchQuery(params, indexName, type, defaultSort, keyMappings, keyMappingsMap);

return elasticsearchTemplate.startScroll(scrollTimeInMillis, searchQuery, Map.class);

}

/**

* 游标查询

* @param scrollId 游标ID

* @param scrollTimeInMillis 游标开启的时间

* @return Page

*/

protected Page commonContinueScroll(String scrollId, long scrollTimeInMillis) {

return elasticsearchTemplate.continueScroll(scrollId, scrollTimeInMillis, Map.class);

}

/**

* 根据游标ID清除游标(提早释放资源,降低ES的负担)

* @param scrollId 游标ID

*/

protected void clearScroll(String scrollId) {

elasticsearchTemplate.clearScroll(scrollId);

}

StoreSearchService中增加游标查询方法以及清除游标方法:

/**

* 游标查询

* @param params 查询条件

* @return page

*/

public Page scroll(Map params) {

IndexConfig config = indexEntity.getConfigByDocCode(DOC_CODE);

// 如果请求参数包含游标ID,则说明执行翻页操作,否则认为开启新的游标查询

String scrollId = params.getOrDefault(SCROLL_ID, null);

if (StringUtils.isNotBlank(scrollId)) {

return commonContinueScroll(params.get(scrollId), config.getScrollTimeInMillis());

}

return commonStartScroll(params, config.getIndexName(), config.getType(), DEFAULT_SORT,

keyMappings, keyMappingsMap, config.getScrollTimeInMillis());

}

public void clearScroll(String scrollId) {

super.clearScroll(scrollId);

}

对外暴露接口:

@PostMapping("/scroll")

public ResponseResult scroll(@RequestBody Map params) {

return ResponseResult.success(storeSearchService.scroll(params));

}

@GetMapping("/scroll/clear/{scrollId}")

public ResponseResult clearScroll(@PathVariable String scrollId) {

storeSearchService.clearScroll(scrollId);

return ResponseResult.success(null);

}

游标查询分为开启和继续两个步骤,接口/scroll中根据_scrollId判断为开启游标查询还是继续游标查询。

若条件允许的话,尽量将游标查询及时关闭,以释放ES集群的资源,降低负担。

源码

Git项目地址:https://github.com/lifengdi/search


版权声明:本文内容由网络用户投稿,版权归原作者所有,本站不拥有其著作权,亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容,请联系我们jiasou666@gmail.com 处理,核实后本网站将在24小时内删除侵权内容。

上一篇:Java 客户端操作 FastDFS 实现文件上传下载替换删除功能
下一篇:Spring boot 整合KAFKA消息队列的示例
相关文章

 发表评论

暂时没有评论,来抢沙发吧~