ElasticSearch使用ElasticsearchTemplate整合spring

1、首先引进maven依赖

<dependency>    
           <groupId>org.springframework.data</groupId>    
            <artifactId>spring-data-elasticsearch</artifactId>    
            <version>2.0.2.RELEASE</version>    
</dependency>  

2.配置es.properties

#默认即为elasticsearch
elasticsearch_cluster_name=elasticsearch
#配置es节点信息，逗号分隔，如果没有指定，则启动ClientNode
elasticsearch_cluster_nodes=172.16.30.56:19300,172.16.30.126:19300

3.配置applicationContext-es.xml，整合spring

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:elasticsearch="http://www.springframework.org/schema/data/elasticsearch"
    xmlns:context="http://www.springframework.org/schema/context"
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
                            http://www.springframework.org/schema/data/elasticsearch http://www.springframework.org/schema/data/elasticsearch/spring-elasticsearch.xsd
                            http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
    <elasticsearch:transport-client id="client"
        cluster-nodes="${elasticsearch_cluster_nodes}" cluster-name="${elasticsearch_cluster_name}" />//对应es.properties中的数据

    <bean name="elasticsearchTemplate"
        class="org.springframework.data.elasticsearch.core.ElasticsearchTemplate">
        <constructor-arg ref="client" />
    </bean>
</beans>

4、导入、注入ElasticsearchTemplate

import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;//导入包
    @Autowired(required=false)
    private ElasticsearchTemplate esTemplate; 

5、新增或者更新数据到ES

    //增加或者更新
    public ReturnResult addEsCashInfo(String cashId, String type) throws Exception {
        ReturnResult result = new ReturnResult();
        if (StringUtils.isBlank(cashId)) {
            String msg = "cashId不能为空";
            logger.info(msg);
            throw new GmallException(msg);
        }
        logger.info("addEsCashInfo cashId=" + cashId + ",type=" + type);
        net.sf.json.JSONObject obj = new net.sf.json.JSONObject();
        if (SuperAppConstant.CASH_CALL_BACK_TYPE.equals(type)) {
            Cash cash = cashMapper.selectByPrimaryKey(cashId);
            obj = net.sf.json.JSONObject.fromObject(cash);
            CashEs cashEs = new CashEs();
            BeanUtils.copyProperties(cashEs, cash);
            IndexQuery indexQuery = new IndexQueryBuilder().withIndexName(SuperAppConstant.ES_INDEX_NAME)
                    .withType(SuperAppConstant.ES_CASH_TYPE).withId(cashId).withObject(cashEs).build();
            esTemplate.index(indexQuery);
        }else{
            String msg = "未知支付类型";
            logger.info(msg);
            throw new GmallException(msg);
        }
        result.setCode(ReturnCodeType.SUCCESS).setMessage("插入成功");
        return result;
}

6、查询ES中的数据

ES查询语句

{
    "query": {
        "bool": {
            "must": [{
                "term": {
                    "state": 1
                }
            }, {
                "term": {
                    "appId": "99999999-9999-9999-9999-999999999999"
                }
            }, {
                "bool": {
                    "should": [{
                        "match": {
                            "payPlatform": {
                                "query": "TL",
                                "type": "boolean"
                            }
                        }
                    }, {
                        "match": {
                            "payPlatform": {
                                "query": "TL_WX_APP",
                                "type": "boolean"
                            }
                        }
                    }, {
                        "match": {
                            "payPlatform": {
                                "query": "TL_ALI",
                                "type": "boolean"
                            }
                        }
                    }, {
                        "match": {
                            "payPlatform": {
                                "query": "TL_WX_JS",
                                "type": "boolean"
                            }
                        }
                    }, {
                        "match": {
                            "payPlatform": {
                                "query": "TL_APP",
                                "type": "boolean"
                            }
                        }
                    }, {
                        "match": {
                            "payPlatform": {
                                "query": "TL_WX_H5",
                                "type": "boolean"
                            }
                        }
                    }, {
                        "match": {
                            "payPlatform": {
                                "query": "WX_GWORLD",
                                "type": "boolean"
                            }
                        }
                    }, {
                        "match": {
                            "payPlatform": {
                                "query": "ALI_GWORLD",
                                "type": "boolean"
                            }
                        }
                    }]
                }
            }]
        }
    }
}

对应使用Java查询

public Sum getEsPaySummaryInfo(String appId) throws Exception {
        SumBuilder sb = AggregationBuilders.sum("tpPrice").field("payPrice");
        BoolQueryBuilder bqb = QueryBuilders.boolQuery();
        bqb.must(QueryBuilders.termQuery("state",SuperAppConstant.PAY_STATUS_SUCCESS));
        bqb.must(QueryBuilders.termQuery("appId",appId));
        bqb.must(QueryBuilders.boolQuery()
                .should(QueryBuilders.matchQuery("payPlatform", SuperAppConstant.PAY_PLATFORM_TL))
                .should(QueryBuilders.matchQuery("payPlatform", SuperAppConstant.PAY_PLATFORM_TL_WX_APP))
                .should(QueryBuilders.matchQuery("payPlatform", SuperAppConstant.PAY_PLATFORM_TL_ALI))
                .should(QueryBuilders.matchQuery("payPlatform", SuperAppConstant.PAY_PLATFORM_TL_WX_JS))
                .should(QueryBuilders.matchQuery("payPlatform", SuperAppConstant.PAY_PLATFORM_TL_APP))
                .should(QueryBuilders.matchQuery("payPlatform", SuperAppConstant.PAY_PLATFORM_TL_WX_H5))
                .should(QueryBuilders.matchQuery("payPlatform", SuperAppConstant.PAY_PLATFORM_WX_GWORLD))
                .should(QueryBuilders.matchQuery("payPlatform", SuperAppConstant.PAY_PLATFORM_ALI_GWORLD))
                );
        SearchQuery searchQuery = new NativeSearchQueryBuilder().withQuery(bqb).withIndices(SuperAppConstant.ES_INDEX_NAME).withTypes(SuperAppConstant.ES_PAY_TYPE)
                .withSearchType(SearchType.DEFAULT)
                .addAggregation(sb).build();
        Aggregations aggregations = esTemplate.query(searchQuery, new ResultsExtractor<Aggregations>() {
            @Override
            public Aggregations extract(SearchResponse response) {
                return response.getAggregations();
            }
        });
        Sum _sum = aggregations.get("tpPrice");
        if(_sum != null){
            logger.info("sum="+_sum.getValue());
        }
        return _sum;
    }

7、聚合查询，分组求和（对应sql的group by）

public Map getEsCashSummaryInfo(String appId) throws Exception {
        Map map = new HashMap<>();
        TermsBuilder tb = AggregationBuilders.terms("cash").field("appId");//appId 是分组字段名，cash是查询结果的别名
        SumBuilder sb = AggregationBuilders.sum("amount").field("paid");//paid是求和字段名称，amount是结果别名
        tb.subAggregation(sb);
        BoolQueryBuilder bqb = QueryBuilders.boolQuery();
        bqb.mustNot(QueryBuilders.termQuery("settled",SuperAppConstant.CASH_STATUS_CANCLED));
        bqb.must(QueryBuilders.termQuery("appId",appId));
        SearchQuery searchQuery = new NativeSearchQueryBuilder().withQuery(bqb).withIndices(SuperAppConstant.ES_INDEX_NAME).withTypes(SuperAppConstant.ES_CASH_TYPE)
                .withSearchType(SearchType.DEFAULT)
                .addAggregation(tb)
                .build();
        Aggregations aggregations = esTemplate.query(searchQuery, new ResultsExtractor<Aggregations>() {
            @Override
            public Aggregations extract(SearchResponse response) {
                return response.getAggregations();
            }
        });
        Terms term = aggregations.get("cash");//获取结果后进行解析
        if(term.getBuckets().size()>0){
            for (Bucket bk : term.getBuckets()) {
                long count = bk.getDocCount();
                //得到所有子聚合  
                Map subaggmap = bk.getAggregations().asMap();  
                //sum值获取方法  
                double amount = ((InternalSum) subaggmap.get("amount")).getValue();  
                map.put("count", count);
                map.put("amount", amount);
            }
            return map;
        }else{
            return null;
        }
    }