springboot项目集成elasticsearch、包括增删改就部分分组、求和、求平均值等计算的用法

简介：ElasticSearch是一个基于Lucene的搜索服务器。它提供了一个分布式多用户能力的全文搜索引擎，基于RESTful web接口。Elasticsearch是用Java开发的，并作为Apache许可条款下的开放源码发布，是当前流行的企业级搜索引擎。设计用于云计算中，能够达到实时搜索，稳定，可靠，快速，安装使用方便。我们建立一个网站或应用程序，并要添加搜索功能，但是想要完成搜索工作的创建是非常困难的。我们希望搜索解决方案要运行速度快，我们希望能有一个零配置和一个完全免费的搜索模式，我们希望能够简单地使用JSON通过HTTP来索引数据，我们希望我们的搜索服务器始终可用，我们希望能够从一台开始并扩展到数百台，我们要实时搜索，我们要简单的多租户，我们希望建立一个云的解决方案。因此我们利用Elasticsearch来解决所有这些问题及可能出现的更多其它问题。

这里使用elasticsearch对用户的登录日志，操作日志等做存储及快速查询，elasticsearch的安装部署就不进行描述，只是展示集成到springboot后的基本操作。

1.导入maven包：

[code]        <!-- springboot集成elasticsearch包 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-elasticsearch</artifactId>
<version>2.0.2.RELEASE</version>
</dependency>

2.实体类，对应索引（数据库）中的类型（数据库表）：

[code]package com.cecjx.manage.demo.domain;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.data.annotation.Id;
import org.springframework.data.elasticsearch.annotations.*;

/**
* 区域数据
*  indexName：索引库名，建议以项目名称命名
*  type：类型，建议以实体类名称命名
*  shard:默认分区数
*  replicas:每个分区默认的备份数
*  refreshInterval:刷新间隔
*  indexStoreType:索引文件存储类型
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
@Document(indexName = "testes1", type = "AreaDO",shards = 1)
@Mapping(mappingPath = "/json/testes-mapping.json")
public class AreaDO {

@Id
@Field(type = FieldType.Keyword)
private String id;
//    @Field(analyzer = "ik", searchAnalyzer = "ik")
private String shortName;//简称
//    @Field(analyzer = "ik_smart", searchAnalyzer = "ik_smart",type = FieldType.Keyword)
private String name;//名称
//    @Field(analyzer = "ik", searchAnalyzer = "ik")
private String mergerName;//全称
private Integer level; //层级 0 1 2 省市区县
private String pinyin;//拼音
private String code; //长途区号
private String zipCode; //邮编
private String first; //首字母
@GeoPointField
private String location;//经纬度
private Integer personCount;//人数，用于统计

}

3.建立索引的映射等，testes-mapping.json放在resource的json目录下：

[code]{
"properties": {
"id": {
"type": "text",
"fielddata":true
},
"mergerName": {
"type": "keyword",
"fields": {
"pinyin": {
"type": "text",
"store": false,
"term_vector": "with_offsets",
"analyzer": "pinyin",
"boost": 10
}
}
},
"name": {
"type": "keyword",
"fields": {
"keyword": {
"type": "text",
"store": false,
"term_vector": "with_offsets",
"analyzer": "ik_smart",
"search_analyzer": "ik_smart",
"boost": 10
}
}
}
}
}

4.elasticsearch的逻辑处理类：

AreaService:

[code]import com.cecjx.common.utils.PageUtils;
import com.cecjx.common.utils.R;
import com.cecjx.manage.demo.domain.AreaDO;

import java.util.Map;

public interface AreaService {
/**
* 新增区域
*
* @param areaDO
* @return
*/
void saveArea(AreaDO areaDO);

/**
* 搜索词搜索，分页返回区域信息
*
* @param params page 当前页码 pageSize 每页大小 search 搜索内容
* @return PageUtils
*/
PageUtils searchAreaPage(Map<String, Object> params);

/**
* 根据id查找
*
* @param id
* @return
*/
AreaDO getAreaById(String id);

void deleteArea(String id);

R count();

//    R deleteAll();
}

AreaServiceImpl:

[code]import com.cecjx.common.utils.PageUtils;
import com.cecjx.common.utils.R;
import com.cecjx.manage.demo.dao.ElasticAreaDao;
import com.cecjx.manage.demo.domain.AreaDO;
import com.cecjx.manage.demo.service.AreaService;
import com.cecjx.manage.demo.util.SnowflakeIdWorker;
import com.github.pagehelper.util.StringUtil;
import org.apache.commons.lang.StringUtils;
import org.elasticsearch.common.unit.Fuzziness;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.StringTerms;
import org.elasticsearch.search.aggregations.metrics.avg.AvgAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.avg.InternalAvg;
import org.elasticsearch.search.aggregations.metrics.max.InternalMax;
import org.elasticsearch.search.aggregations.metrics.max.MaxAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.min.InternalMin;
import org.elasticsearch.search.aggregations.metrics.min.MinAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.stats.InternalStats;
import org.elasticsearch.search.aggregations.metrics.stats.StatsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.sum.InternalSum;
import org.elasticsearch.search.aggregations.metrics.sum.Sum;
import org.elasticsearch.search.aggregations.metrics.sum.SumAggregationBuilder;
import org.elasticsearch.search.aggregations.support.ValuesSourceAggregationBuilder;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.data.elasticsearch.core.aggregation.AggregatedPage;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.SearchQuery;
import org.springframework.stereotype.Service;

import java.util.*;

/**
* elasticsearch逻辑处理
*
* @author huangquanguang
* @create 2020/01/02 10:20
*/
@Service("areaServiceImpl")
public class AreaServiceImpl implements AreaService {
private static final Logger logger = LoggerFactory.getLogger(AreaServiceImpl.class);

@Autowired
private ElasticAreaDao elasticAreaDao;
@Autowired
private ElasticsearchTemplate elasticsearchTemplate;

/**
* 新增修改通用，有id则修改，没有则新增
*
* @param areaDO
*/
@Override
public void saveArea(AreaDO areaDO) {
String oldId = areaDO.getId();
if (StringUtil.isEmpty(oldId)) {
SnowflakeIdWorker idWorker = new SnowflakeIdWorker(0, 0);
Long id = idWorker.nextId();
areaDO.setId(id.toString());
}
elasticAreaDao.save(areaDO);
}

/**
* 根据id查询
*
* @param id
* @return
*/
@Override
public AreaDO getAreaById(String id) {
Optional<AreaDO> area = elasticAreaDao.findById(id);
AreaDO areaDO = area.get();
return areaDO;
}

/**
* 根据id删除
*
* @param id
*/
@Override
public void deleteArea(String id) {
elasticAreaDao.deleteById(id);
}

/**
* @author  huangquanguang
* @date  2020/1/16 10:24
* @param
* @return
* es的统计
*/
@Override
public R count() {

try {
/**
* 求和，统计各省人数总和
*/
QueryBuilder queryBuilder = QueryBuilders.boolQuery()
.must(QueryBuilders.rangeQuery("personCount"));
// 聚合查询。personCount是要统计的字段，sum_codes是自定义的别名
SumAggregationBuilder sumBuilder = AggregationBuilders.sum("sum_codes").field("personCount");
SearchQuery searchQuery = new NativeSearchQueryBuilder()
.withQuery(queryBuilder)
.addAggregation(sumBuilder)
.build();
double personAmount = elasticsearchTemplate.query(searchQuery, response -> {
InternalSum sum = (InternalSum) response.getAggregations().asList().get(0);
return sum.getValue();
});
logger.info("人数总和为:" + personAmount);

/**
* 统计某个字段(name=广东)的数量
*/
QueryBuilder queryBuilderName = QueryBuilders.matchPhrasePrefixQuery("name", "广东").slop(0);
ValuesSourceAggregationBuilder vsb = AggregationBuilders.terms("count").field("name");
SearchQuery searchQueryName = new NativeSearchQueryBuilder()
.withQuery(queryBuilderName)
.addAggregation(vsb)
.build();
long nameAmount = elasticsearchTemplate.query(searchQueryName, response -> {
StringTerms stringTerms = (StringTerms) response.getAggregations().asList().get(0);
//获得所有的桶
List<StringTerms.Bucket> buckets = stringTerms.getBuckets();
long count = buckets.get(0).getDocCount();
return count;
});
logger.info("name=广东的数量为:" + nameAmount);

/**
* 按某个字段分组求和（这里按照省份分组统计人数总和）
* 实现sql:select field1, field2, sum(field3) from table_name group by field1, field2;
*/
NativeSearchQueryBuilder queryBuilderGroup = new NativeSearchQueryBuilder();
// 聚合查询。name是要统计的字段，group_name是自定义的别名
queryBuilderGroup.addAggregation(AggregationBuilders.terms("groupName").field("name")
.subAggregation(AggregationBuilders.sum("sum_codes").field("personCount")));
AggregatedPage<AreaDO> result = elasticsearchTemplate.queryForPage(queryBuilderGroup.build(), AreaDO.class);
//解析聚合
Aggregations aggregations = result.getAggregations();
//获取指定名称的聚合
StringTerms terms = aggregations.get("groupName");
//获取桶
List<StringTerms.Bucket> buckets = terms.getBuckets();
//遍历打印
List list = new ArrayList();
for (StringTerms.Bucket bucket : buckets) {
Sum sum = bucket.getAggregations().get("sum_codes");
Map map = new HashMap<>();
logger.info("省：" + bucket.getKeyAsString());
logger.info("记录行数：" + bucket.getDocCount());
logger.info("总人数统计：" + sum.getValue());
map.put("省", bucket.getKeyAsString());
map.put("记录行数", bucket.getDocCount());
map.put("总人数统计", sum.getValue());
list.add(map);
}

/**
* 求平均，统计各省人数平均值
*/
QueryBuilder queryBuilderAvg = QueryBuilders.boolQuery()
.must(QueryBuilders.rangeQuery("personCount"));
// 聚合查询。personCount是要统计的字段，sum_codes是自定义的别名
AvgAggregationBuilder ab = AggregationBuilders.avg("avg_count").field("personCount");
SearchQuery searchQueryAvg = new NativeSearchQueryBuilder()
.withQuery(queryBuilderAvg)
.addAggregation(ab)
.build();
double avgAmount = elasticsearchTemplate.query(searchQueryAvg, response -> {
InternalAvg avg = (InternalAvg) response.getAggregations().asList().get(0);
return avg.getValue();
});
logger.info("人数平均值为:" + avgAmount);

/**
* 求最大值（人数最多的省份）
*/
BoolQueryBuilder builder = QueryBuilders.boolQuery();
FieldSortBuilder sort = SortBuilders.fieldSort("personCount").order(SortOrder.DESC);
Page<AreaDO> areaPage = elasticAreaDao.search(new NativeSearchQueryBuilder()
.withQuery(builder)
.withSort(sort)
.build());
List<AreaDO> areaDOS = areaPage.getContent();
logger.info("人数最多的省份为：" + areaDOS.get(0).getName());

QueryBuilder queryBuilderMax = QueryBuilders.boolQuery()
.must(QueryBuilders.rangeQuery("personCount"));
// 聚合查询。personCount是要统计的字段，sum_codes是自定义的别名
MaxAggregationBuilder mb = AggregationBuilders.max("max_count").field("personCount");
SearchQuery searchQueryMax = new NativeSearchQueryBuilder()
.withQuery(queryBuilderMax)
.addAggregation(mb)
.build();
double maxAmount = elasticsearchTemplate.query(searchQueryMax, response -> {
InternalMax max = (InternalMax) response.getAggregations().asList().get(0);
return max.getValue();
});
logger.info("最多人的省份人数为:" + maxAmount);

/**
* 求最小值（人数最少的省份）
*/
BoolQueryBuilder builderMin = QueryBuilders.boolQuery();
FieldSortBuilder sortMin = SortBuilders.fieldSort("personCount").order(SortOrder.ASC);
Page<AreaDO> areaPageMin = elasticAreaDao.search(new NativeSearchQueryBuilder()
.withQuery(builderMin)
.withSort(sortMin)
.build());
List<AreaDO> areasMin = areaPageMin.getContent();
logger.info("人数最少的省份为：" + areasMin.get(0).getName());
QueryBuilder queryBuilderMin = QueryBuilders.boolQuery()
.must(QueryBuilders.rangeQuery("personCount"));
// 聚合查询。personCount是要统计的字段，sum_codes是自定义的别名
MinAggregationBuilder min = AggregationBuilders.min("max_count").field("personCount");
SearchQuery searchQueryMin = new NativeSearchQueryBuilder()
.withQuery(queryBuilderMin)
.addAggregation(min)
.build();
double minAmount = elasticsearchTemplate.query(searchQueryMin, response -> {
InternalMin internalMin = (InternalMin) response.getAggregations().asList().get(0);
return internalMin.getValue();
});
logger.info("最少人的省份人数为:" + minAmount);

//多种聚合结果
QueryBuilder queryBuilderStats = QueryBuilders.boolQuery()
.must(QueryBuilders.rangeQuery("personCount"));
StatsAggregationBuilder stats = AggregationBuilders.stats("personStats").field("personCount");
SearchQuery searchQueryStats = new NativeSearchQueryBuilder()
.withQuery(queryBuilderStats)
.addAggregation(stats)
.build();
InternalStats statsAmount = elasticsearchTemplate.query(searchQueryStats, response -> {
InternalStats internalStats = (InternalStats) response.getAggregations().asList().get(0);
logger.info("最多人的省份"+internalStats.getMaxAsString());
logger.info("最多人的省份的人数"+internalStats.getMax());
return internalStats;
});
return R.ok().put("人数总和为", personAmount).put("name=广东的数量为", nameAmount)
.put("省份分组求人数和", list).put("人数平均值为", avgAmount).put("人数最多的省份为", areaDOS.get(0).getName())
.put("最多人的省份人数为", maxAmount).put("最少人的省份为", areasMin.get(0).getName())
.put("最少人的省份人数为", minAmount);

} catch (Exception e) {
e.printStackTrace();
return R.error();
}

//        //（3）聚合过滤
//        FilterAggregationBuilder fab = AggregationBuilders.filter("uid_filter").filter(QueryBuilders.queryStringQuery("uid:001"));
//        //（4）按某个字段分组
//        TermsAggregationBuilder tb = AggregationBuilders.terms("group_name").field("name");
//        //（5）求和
//        SumAggregationBuilder sumBuilder = AggregationBuilders.sum("sum_price").field("price");
//        //（6）求平均
//        AvgAggregationBuilder ab = AggregationBuilders.avg("avg_price").field("price");
//        //（7）求最大值
//        MaxAggregationBuilder mb = AggregationBuilders.max("max_price").field("price");
//        //（8）求最小值
//        MinAggregationBuilder min = AggregationBuilders.min("min_price").field("price");
//        //（9）按日期间隔分组
//        DateHistogramAggregationBuilder dhb = AggregationBuilders.dateHistogram("dh").field("date");
//        //（10）获取聚合里面的结果
//        TopHitsAggregationBuilder thb = AggregationBuilders.topHits("top_result");
//        //（11）嵌套的聚合
//        NestedAggregationBuilder nb = AggregationBuilders.nested("negsted_path").path("quests");
//        //（12）反转嵌套
//        ReverseNestedAggregationBuilder reb = AggregationBuilders.reverseNested("res_negsted").path("kps ");

}

/**
* 删除所有
* @return
*/
//    @Override
//    public R deleteAll() {
//        try {
//            elasticAreaRepository.deleteAll();
//            return R.ok();
//        } catch (Exception e) {
//            e.printStackTrace();
//            return R.error("系统异常");
//        }
//    }

/**
* 分页查询
*
* @param params page 当前页码 pageSize 每页大小 search 搜索内容
* @return
*/
@Override
public PageUtils searchAreaPage(Map<String, Object> params) {
int page = Integer.parseInt((String) params.get("page")) - 1;
int pageSize = Integer.parseInt((String) params.getOrDefault("pageSize", 10));
//        String sort = (String) params.get("sort");
String name = (String) params.get("name");
String mergerName = (String) params.get("mergerName");
// 构建搜索查询
SearchQuery searchQuery = getLogSearchQuery(page, pageSize, name,mergerName);
logger.info("searchLogPage: searchContent [{}] \n DSL  = \n {}", name, searchQuery.getQuery().toString());
Page<AreaDO> areaPage = elasticAreaDao.search(searchQuery);
List<AreaDO> areaDOS = areaPage.getContent();
return new PageUtils(areaDOS, areaPage.getTotalElements());
}

/**
* 根据搜索词构造搜索查询语句
* 代码流程：
* - 精确查询
* - 模糊查询
* - 排序查询
* - 设置分页参数
*
* @param pageNumber    当前页码
* @param pageSize      每页大小
* @param name 搜索内容
* @return
*/
private SearchQuery getLogSearchQuery(Integer pageNumber, Integer pageSize, String name,String mergerName) {
//创建builder
BoolQueryBuilder builder = QueryBuilders.boolQuery();
/**
*  must
所有的语句都 必须（must） 匹配，与 AND 等价。
must_not
所有的语句都 不能（must not） 匹配，与 NOT 等价。
should
至少有一个语句要匹配，与 OR 等价。
trem
精确查找 与= 号等价。
match
模糊匹配 与like 等价。
*/
//设置多字段组合
if (StringUtils.isNotBlank(name)) {
//分词查询，采用默认的分词器(这里设置为ik_smart)
//            builder.must(QueryBuilders.multiMatchQuery(name, "name", "mergerName"));
//模糊搜索
builder.must(QueryBuilders.fuzzyQuery("name", name).fuzziness(Fuzziness.ONE));
//精确搜索
//            builder.must(QueryBuilders.termQuery("name", searchContent));
//分词查询，采用默认的分词器
//            QueryBuilder queryBuilder2 = QueryBuilders.matchQuery("name", name);
//不分词搜索
//            builder.must(QueryBuilders.matchPhrasePrefixQuery("name", name).slop(0));
}
if (StringUtils.isNotBlank(mergerName)) {
builder.must(QueryBuilders.queryStringQuery(mergerName).field("mergerName"));//左右模糊;
}
//设置排序
FieldSortBuilder sort = SortBuilders.fieldSort("id").order(SortOrder.DESC);
//设置分页 此处升级
Pageable pageable = PageRequest.of(pageNumber, pageSize);
return new NativeSearchQueryBuilder()
.withPageable(pageable)
.withQuery(builder)
//                .withSort(sort)
.build();
}

}

5.ElasticAreaDao：

[code]import com.cecjx.manage.demo.domain.AreaDO;
import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;

import java.util.Optional;

public interface ElasticAreaDao extends ElasticsearchRepository<AreaDO,String> {

Optional<AreaDO> findById(String id);

void deleteById(String id);
}

6.AreaController:

[code]import com.cecjx.common.utils.PageUtils;
import com.cecjx.common.utils.R;
import com.cecjx.manage.demo.domain.AreaDO;
import com.cecjx.manage.demo.service.AreaService;
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;
import java.util.Map;

/**
* elasticsearch增删改查某个索引demo
*
* @author huangquanguang
* @create 2020/01/02 10:20
*/
@Controller
@RequestMapping(value = "elasticsearch")
public class AreaController {

@Resource(name = "areaServiceImpl")
private AreaService areaService;

@GetMapping(value = "index")
public String index() {
return "manage/demo/elasticsearch/area";
}

/**
* 新增页
*
* @return
*/
@RequestMapping(value = "/add")
public String add() {
return "manage/demo/elasticsearch/areaAdd";
}

/**
* 修改页
*
* @return
*/
@GetMapping("/edit/{id}")
public String edit(@PathVariable("id") String id, Model model) {
AreaDO areaDO = areaService.getAreaById(id);
model.addAttribute("areaDO", areaDO);
return "manage/demo/elasticsearch/areaEdit";
}

/**
* 根据id删除
*
* @param id
*/
@ResponseBody
@PostMapping("/remove")
public R deletePicture(String id) {
try {
areaService.deleteArea(id);
return R.ok();
} catch (Exception e) {
e.printStackTrace();
return R.error();
}
}

/**
* 批量删除
*
* @param ids
* @return
*/
@PostMapping("/batchRemove")
@ResponseBody
R batchRemove(@RequestParam("ids[]") String[] ids) {
try {
for (String str : ids) {
areaService.deleteArea(str);
}
return R.ok();
} catch (Exception e) {
e.printStackTrace();
return R.error();
}
}

/**
* 这个方法被新增和修改请求共用,如果是新增，id必须为空，否则是修改请求
*
* @param areaDO
* @return
*/
@ResponseBody
@RequestMapping(value = "/save", method = RequestMethod.POST)
public R saveArea(AreaDO areaDO) {
try {
areaService.saveArea(areaDO);
return R.ok();
} catch (Exception e) {
e.printStackTrace();
return R.error("系统异常");
}
}

/**
* 分页查询
*
* @param params
* @return
*/
@RequestMapping(value = "/list", method = RequestMethod.GET)
@ResponseBody
public PageUtils list(@RequestParam Map<String, Object> params) {
return areaService.searchAreaPage(params);
}

/**
* 删除所有
* @return
*/
//    @RequestMapping(value = "/deleteAll", method = RequestMethod.GET)
//    @ResponseBody
//    public R deleteAll() {
//        return areaService.deleteAll();
//    }

/**
* 统计例子
*
* 1-统计某个字段的数量
* 2-去重统计某个字段的数量（有少量误差）
* 3-聚合过滤
* 4-按某个字段分组
* 5-求和
* 6-求平均
* 7-求最大值
* 8-求最小值
* 9-按日期间隔分组
* 10-获取聚合里面的结果
* 11-嵌套的聚合
* 12-反转聚合
* @return
*/
@RequestMapping(value = "/count", method = RequestMethod.GET)
@ResponseBody
public R count() {
return areaService.count();
}
}

7.注意：这里的增删改查分页已经调整好跟bootdo的前端使用一致，因此前端代码就不贴了。直接就是使用bootstrapTable的。