前言
但想要用ElasticSearch 在海量的数据中找到和自己相关的业务数据，实现对已有的数据实现全文检索、分类统计等功能并应用到业务系统中，还需要借助Java API来实现，今天总结一下ElasticSearch Java API的简单使用。
一：获取ElasticSearch client连接对象
public Client getEsClient(){
Client client = null;
try{
Settings settings = ImmutableSettings.settingsBuilder()
.put("client.transport.sniff", true) //自动嗅探整个ES集群节点
.put("client", true)
.put("data",false)
.put("cluster.name","elasticsearch") //设置集群名字
.put("number_of_shards", 2) //2个主分片
.put("index.refresh_interval", "5s") //每5秒提交一次数据，类似oracle中的commit
.build();
client = new TransportClient(settings).addTransportAddress(new InetSocketTransportAddress("hostName", 9200));
}catch(Exception ex){
ex.printStackTrace();
}
return client;
}
如果觉得这种获取连接方式麻烦还可以直接使用：TransportClient类来进行连接，TransportClient实现了Client接口。
二：List查询方法
List<Map<String, Object>> dataList = new ArrayList<Map<String, Object>>();
Client client = null;
try{
client = getEsClient();
//设置查询条件（类似sql中 where = eventid )
QueryBuilder qb = QueryBuilders.boolQuery().must(QueryBuilders.termQuery("eventid", eventid));
// 设置ES中 indexName 和 indexType，以及QueryBuilders 对象
SearchRequestBuilder requestBuilder = client.prepareSearch(EventBean.indexName)
.setTypes(EventBean.indexType)（添加类型）
.setQuery(qb);
//执行查询 （可以设置排序、分页）
SearchResponse actionGet = requestBuilder.addSort(SortBuilders.fieldSort("occurtime").order(SortOrder.DESC))
.setFrom((pageNum-1) * pageSize)
.setSize(pageNum * pageSize)
.execute().actionGet();
//遍历查询结果
if(actionGet != null){
SearchHits hits = actionGet.getHits();
if(hits != null && hits.getHits() != null){
Map<String, Object> hitMap = null;
for (SearchHit hit : hits.getHits()){
hitMap = hit.getSource();
if(hitMap == null || hitMap.size() <= 0){
continue;
}
dataList.add(hitMap );
}
}
}
}catch(Exception ex){
ex.printStackTrace();
}finally{
//关闭client
}
return dataList ;
三：添加方法
boolean optFlag = false;
Client client = null;
try{
//获取Elasticsearch client对象
client = getEsClient();
//构造请求对象
BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
IndexRequestBuilder indexRequestBuilder = null;
//根据请求对象bodyList, 组织bulkRequestBuilder对象
String indexId = "";
Map<String, Object> bodyMap = null;
for (int count = 0; count < bodyList.size(); count++) {
bodyMap = bodyList.get(count);
if (bodyMap == null) {
break;
}
//组织索引对象id
indexId = ConverUtils.Obj2Str(bodyMap.get("id"), "");
indexRequestBuilder = client.prepareIndex(indexName, indexType).setId(indexId).setSource(bodyMap);
bulkRequestBuilder.add(indexRequestBuilder);
}
//批量更新执行
BulkResponse bulkResponse = bulkRequestBuilder.execute().actionGet();
if (bulkResponse.hasFailures()) {
optFlag = false;
} else {
optFlag = true;
}
}catch(Exception ex){
ex.printStackTrace();
}finally{
//关闭client
}
return optFlag ;
四：更新方法
boolean optFlag = false;
Client client = null;
try{
//获取Elasticsearch client对象
client = getEsClient();
//构造请求对象
BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
UpdateRequestBuilder updateRequestBuilder = null;
//根据请求对象cvdMap，组织索引对象
String indexId = getEsIndexId(client, ConverUtils.Obj2long(cvdMap.get("id"), 0L));
//获取索引(getEsIndexId)
updateRequestBuilder = client.prepareUpdate(indexName, indexType, indexId).setDoc(cvdMap);
bulkRequestBuilder.add(updateRequestBuilder);
//批量更新执行
BulkResponse bulkResponse = bulkRequestBuilder.setRefresh(true).execute().actionGet();
if (bulkResponse.hasFailures()) {
optFlag = false;
} else {
optFlag = true;
}
}catch(Exception ex){
ex.printStackTrace();
}finally{
//关闭client
}
return optFlag ;
五：删除
boolean optFlag = false;
Client client = null;
try{
//获取Elasticsearch client对象
client = getEsClient();
//根据id_array数组， 组织索引对象
BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
DeleteRequestBuilder deleteRequestBuilder = null;
for(int i=0; i<id_array.length; i++){
if(id_array[i] == null || id_array[i].trim().equals("")){
continue;
}
//根据indexName, indexType, indexId进行删除(这里nfa_dictionary索引中ID字段值和ES自有_id值相同， 可以操作)
deleteRequestBuilder = client.prepareDelete(indexName, indexType, id_array[i]);
bulkRequestBuilder.add(deleteRequestBuilder.request());
}
//进行批量删除操作
BulkResponse bulkResponse = bulkRequestBuilder.execute().actionGet();
if (bulkResponse.hasFailures()) {
optFlag = false;
}else {
optFlag = true;
}
}catch(Exception ex){
ex.printStackTrace();
}finally{
//关闭client
}
return optFlag ;
好啦，这就是Java操作ES的基本增删改查，是不是跟操作数据库很相似呢？喜欢的话关注一下吧。