ElasticSearch简单使用

What is ElasticSearch ？

Elasticsearch是一个基于Apache Lucene(TM)的开源搜索引擎。无论在开源还是专有领域，Lucene可以被认为是迄今为止最先进、性能最好的、功能最全的搜索引擎库。

但是，Lucene只是一个库。想要使用它，你必须使用Java来作为开发语言并将其直接集成到你的应用中，更糟糕的是，Lucene非常复杂，你需要深入了解检索的相关知识来理解它是如何工作的。

Elasticsearch也使用Java开发并使用Lucene作为其核心来实现所有索引和搜索的功能，但是它的目的是通过简单的RESTful API来隐藏Lucene的复杂性，从而让全文搜索变得简单。

不过，Elasticsearch不仅仅是Lucene和全文搜索，我们还能这样去描述它：

分布式的实时文件存储，每个字段都被索引并可被搜索

分布式的实时分析搜索引擎

可以扩展到上百台服务器，处理PB级结构化或非结构化数据

而且，所有的这些功能被集成到一个服务里面，你的应用可以通过简单的RESTful API、各种语言的客户端甚至命令行与之交互。

上手Elasticsearch非常容易。它提供了许多合理的缺省值，并对初学者隐藏了复杂的搜索引擎理论。它开箱即用（安装即可使用），只需很少的学习既可在生产环境中使用。

Elasticsearch在Apache 2 license下许可使用，可以免费下载、使用和修改。

随着你对Elasticsearch的理解加深，你可以根据不同的问题领域定制Elasticsearch的高级特性，这一切都是可配置的，并且配置非常灵活。

Config elasticSearch.properties

#配置elasticsearch index
csg_bill_index_name=hsc_test_index
#配置elasticsearch type
csg_bill_type_name=billdetails
#配置elasticsearch es address
csg_bill_es_address=l-test.h.beta.cn0:9300
#配置elasticsearch cluster
csg_bill_es_cluster=elasticsearch_f

Create elasticSearch TransportSessionFactory

/**
* Created by xueping.you on 15-7-30.
*/
@Service
public class ESTransportSessionFactory  {

private final static Logger LOGGER = LoggerFactory.getLogger(ESTransportSessionFactory.class);

@Value("${csg_bill_es_address}")
private String elasticAddress;

@Value("${csg_bill_es_cluster}")
private String elasticCluster;

private TransportClient transportClient;

@PostConstruct
public void init(){
checkArgument(!Strings.isEmpty(elasticCluster) ,
"ElasticSearch Cluster Name Null or EmptyString");
TransportAddress[] transportAddresses = null;
List<String> addressList =
Lists.newArrayList(
Splitter.on(",").trimResults().omitEmptyStrings().split(elasticAddress)
);
checkArgument(!CollectionUtils.isEmpty(addressList) ,
"ElasticSearch Cluster Address Can't be Empty");
transportAddresses = new TransportAddress[addressList.size()];
Splitter splitter = Splitter.on(":").omitEmptyStrings().trimResults();
for(int i=0; i<addressList.size() ; i++){
List<String> singleAddressPair =
Lists.newArrayList(splitter.split(addressList.get(i)));
checkArgument(singleAddressPair.size()==2 ,
"ElasticSearch Address format address:port error" + addressList.get(i)
);
transportAddresses[i] = new InetSocketTransportAddress(
singleAddressPair.get(0),
Integer.parseInt(singleAddressPair.get(1))
);
}
Settings settings = ImmutableSettings.builder()
.put("cluster.name" , elasticCluster)
.build();
transportClient = new TransportClient(settings);
transportClient.addTransportAddresses(transportAddresses);

LOGGER.info("ElasticSearch Init Done ElasticAddress={} , ElasticClusterName={}"
, elasticAddress , elasticCluster);
}

@PreDestroy
public void destroy() {
if (transportClient != null) {
transportClient.close();
}
}

public TransportClient getTransportClient() {
return transportClient;
}

}

Service Method Interface

/**
* Created by xueping.you on 15-7-30.
*/
@Service
public class ESBillDetailServiceBase {
private final static Logger LOGGER = LoggerFactory.getLogger(ESBillDetailServiceBase.class);

@Resource
private ESTransportSessionFactory esTransportSessionFactory;

@Value("${csg_bill_index_name}")
private String ES_INDEX;

@Value("${csg_bill_type_name}")
private String ES_TYPE;

private static ObjectMapper objectMapper = new ObjectMapper();

static {
objectMapper.setDateFormat(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));
}

/**
* Method1 将订单号 , 对账单号 , 产品Id , 账单号 ,结算供应商ID,结算对象Id,create_ts进行索引
* @param esBillDetails
*/
public void insertAllOnDuplicateIdCover(List<ESBillDetail> esBillDetails){
Stopwatch stopwatch = Stopwatch.createStarted();
if(CollectionUtils.isEmpty(esBillDetails)){
return;
}

TransportClient client = esTransportSessionFactory.getTransportClient();
BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();

for(ESBillDetail esBillDetail : esBillDetails){
String source;
try {
source = objectMapper.writeValueAsString(esBillDetail);
} catch (Exception e1) {
LOGGER.error("create billDetail index error source : " + esBillDetail, e1);
QMonitor.recordOne(QMonitorConstants.ES_BILL_DETAIL_INDEXING_ERROR);
continue;
}
IndexRequestBuilder indexRequestBuilder =
client.prepareIndex(ES_INDEX, ES_TYPE,
esBillDetail.getId()).setSource(source);
bulkRequestBuilder.add(indexRequestBuilder);
}

BulkResponse bulkResponse = bulkRequestBuilder.execute().actionGet();
if(bulkResponse.hasFailures()){
Iterator<BulkItemResponse> itemResponseIterator = bulkResponse.iterator();
while(itemResponseIterator.hasNext()){
BulkItemResponse itemResponse = itemResponseIterator.next();
if(itemResponse!=null && itemResponse.isFailed()){
LOGGER.info("indexing billDetail error indexMessage={},
errorMassage={}",Joiner.on(',')                     .join(itemResponse.getIndex(),itemResponse.getType(),itemResponse.getId()),
JsonUtils.toJSONString(itemResponse));
}
}
QMonitor.recordOne(QMonitorConstants.ES_BILL_DETAIL_INDEXING_ERROR);
}

QMonitor.recordOne(QMonitorConstants.ES_BILL_DETAIL_INDEXING_DONE , stopwatch.elapsed(TimeUnit.MILLISECONDS));
}

/**
* Method2 通过订单号 ，对账单号, 产品Id , 账单号 的任意条件进行查询，支持分页
*/
public List<ESBillDetail> query(ESBillDetailQuery esBillDetailQuery , Pagination pagination){
try {
Stopwatch stopwatch = Stopwatch.createStarted();
List<ESBillDetail> esBillDetails = Lists.newArrayList();
BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
for(Map.Entry entry : esBillDetailQuery.entrySet()){
ESBillDetailQuery.Param param = (ESBillDetailQuery.Param)entry.getKey();
param.decorateBoolQueryBuilder(entry.getValue(),boolQueryBuilder);
}
SearchRequestBuilder requestBuilder = esTransportSessionFactory.getTransportClient()
.prepareSearch(ES_INDEX)
.setTypes(ES_TYPE)
.setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
.setQuery(boolQueryBuilder);
if(pagination!=null){
requestBuilder.setFrom(pagination.getStartIndex())
.setSize(pagination.getPageSize());
}
SearchResponse searchResponse = requestBuilder.execute().actionGet();
SearchHits searchHits = searchResponse.getHits();
for(SearchHit searchHit : searchHits){
Map<String , Object> map = searchHit.getSource();
ESBillDetail esBillDetail = objectMapper.readValue(
JsonUtils.toJSONString(map) ,
new TypeReference<ESBillDetail>() {});
esBillDetails.add(esBillDetail);
}
LOGGER.info("Query ESDetail done query={} , time={}" , esBillDetailQuery , stopwatch.elapsed(TimeUnit.MILLISECONDS));
QMonitor.recordOne(QMonitorConstants.ES_BILL_QUERY , stopwatch.elapsed(TimeUnit.MILLISECONDS));
return esBillDetails;
}catch (Exception e){
LOGGER.error("elastic search query Error query={} , page={}" , esBillDetailQuery , pagination , e);
QMonitor.recordOne(QMonitorConstants.ES_BILL_DETAIL_QUERY_ERROR);
throw new RuntimeException("ES 查询异常");
}
}

/**
* 此方法慎用！！！！！！！！！！！！！！！！！！！
* Method3 通过匹配条件删除ES 中的记录
*/
public boolean delete(ESBillDetailQuery esBillDetailQuery){
try {
Stopwatch stopwatch = Stopwatch.createStarted();
BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
for(Map.Entry entry : esBillDetailQuery.entrySet()){
ESBillDetailQuery.Param param = (ESBillDetailQuery.Param)entry.getKey();
param.decorateBoolQueryBuilder(entry.getValue(),boolQueryBuilder);
}
DeleteByQueryResponse deleteByQueryResponse = esTransportSessionFactory.getTransportClient()
.prepareDeleteByQuery(ES_INDEX )
.setTypes(ES_TYPE)
.setQuery(boolQueryBuilder).execute().actionGet();
LOGGER.info("Delete ESDetail done query={} , time={}" , esBillDetailQuery , stopwatch.elapsed(TimeUnit.MILLISECONDS));
QMonitor.recordOne(QMonitorConstants.ES_BILL_DELETE , stopwatch.elapsed(TimeUnit.MILLISECONDS));
return deleteByQueryResponse.status().equals(RestStatus.OK) ? true : false;
}catch (Exception e){
LOGGER.error("elastic search delete Error query={} , page={}" , esBillDetailQuery , e);
QMonitor.recordOne(QMonitorConstants.ES_BILL_DETAIL_QUERY_ERROR);
throw new RuntimeException("ES 删除记录异常");
}
}

ESBillDetailQuery

/**
* Created by xueping.you on 15-7-30.
*/
public class ESBillDetailQuery extends GenericQuery<ESBillDetailQuery.Param> {

//全为Str类型
public enum Param{
ids {
@Override
public <T, Q extends QueryBuilder> void decorateBoolQueryBuilder(T param, Q queryBuilder) {
List<String> strsParam = (List<String>)param;
BoolQueryBuilder boolQueryBuilder = (BoolQueryBuilder)queryBuilder;
if(CollectionUtils.isEmpty(strsParam)){
return;
}
BoolQueryBuilder subCodeQuery = QueryBuilders.boolQuery();
for(String strParam : strsParam){
subCodeQuery.should(QueryBuilders.matchQuery("billNo" , strParam).operator(MatchQueryBuilder.Operator.AND));
}
boolQueryBuilder.must(subCodeQuery);
}
},
billNos {
@Override
public <T, Q extends QueryBuilder> void decorateBoolQueryBuilder(T param, Q queryBuilder) {
List<String> strsParam = (List<String>)param;
BoolQueryBuilder boolQueryBuilder = (BoolQueryBuilder)queryBuilder;
if(CollectionUtils.isEmpty(strsParam)){
return;
}
BoolQueryBuilder subCodeQuery = QueryBuilders.boolQuery();
for(String strParam : strsParam){
subCodeQuery.should(QueryBuilders.matchQuery("billNo" , strParam).operator(MatchQueryBuilder.Operator.AND));
}
boolQueryBuilder.must(subCodeQuery);
}
},
orderNos {
@Override
public <T, Q extends QueryBuilder> void decorateBoolQueryBuilder(T param, Q queryBuilder) {
List<String> strsParam = (List<String>)param;
BoolQueryBuilder boolQueryBuilder = (BoolQueryBuilder)queryBuilder;
if(CollectionUtils.isEmpty(strsParam)){
return;
}
BoolQueryBuilder subCodeQuery = QueryBuilders.boolQuery();
for(String strParam : strsParam){
subCodeQuery.should(QueryBuilders.matchQuery("orderNo" , strParam).operator(MatchQueryBuilder.Operator.AND));
}
boolQueryBuilder.must(subCodeQuery);
}
},
productIds {
@Override
public <T, Q extends QueryBuilder> void decorateBoolQueryBuilder(T param, Q queryBuilder) {
List<String> strsParam = (List<String>)param;
BoolQueryBuilder boolQueryBuilder = (BoolQueryBuilder)queryBuilder;
if(CollectionUtils.isEmpty(strsParam)){
return;
}
BoolQueryBuilder subCodeQuery = QueryBuilders.boolQuery();
for(String strParam : strsParam){
subCodeQuery.should(QueryBuilders.matchQuery("productId" , strParam).operator(MatchQueryBuilder.Operator.AND));
}
boolQueryBuilder.must(subCodeQuery);
}
},
reconcileIds {
@Override
public <T, Q extends QueryBuilder> void decorateBoolQueryBuilder(T param, Q queryBuilder) {
List<String> strsParam = (List<String>)param;
BoolQueryBuilder boolQueryBuilder = (BoolQueryBuilder)queryBuilder;
if(CollectionUtils.isEmpty(strsParam)){
return;
}
BoolQueryBuilder subCodeQuery = QueryBuilders.boolQuery();
for(String strParam : strsParam){
subCodeQuery.should(QueryBuilders.matchQuery("reconcileId" , strParam).operator(MatchQueryBuilder.Operator.AND));
}
boolQueryBuilder.must(subCodeQuery);
}
};

public abstract <T, Q extends QueryBuilder> void decorateBoolQueryBuilder(T param ,  Q queryBuilder );
}

public ESBillDetailQuery() {
}

public ESBillDetailQuery(Integer startIndex, Integer maxCount) {
super(startIndex, maxCount);
}

public ESBillDetailQuery(Class<Param> paramType) {
super(paramType);
}
}