前言

但想要用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的基本增删改查，是不是跟操作数据库很相似呢？喜欢的话关注一下吧。