package com.mdl.monitor.init;

import com.google.gson.GsonBuilder;

import io.searchbox.client.JestClient;

import io.searchbox.client.JestClientFactory;

import io.searchbox.client.config.HttpClientConfig;

/**

* 初始化es

*/

public class InitElasticSearchConfig {

private JestClient client ;

public JestClient getClient() {

return client;

}

public InitElasticSearchConfig(String esUrl){

client = getClientConfig(esUrl) ;

}

public JestClient getClientConfig(String esUrl){

JestClientFactory factory = new JestClientFactory();

factory.setHttpClientConfig(new HttpClientConfig

.Builder(esUrl)

.gson(new GsonBuilder().setDateFormat("yyyy-MM-dd'T'HH:mm:ss").create())

.multiThreaded(true)

.readTimeout(10000)

.build());

JestClient client = factory.getObject();

return client ;

}

}

package com.mdl.monitor.repositorys.elasticsearch;

import java.util.List;

import com.google.gson.JsonObject;

import io.searchbox.client.JestResult;

import io.searchbox.core.SearchResult.Hit;

import io.searchbox.core.SuggestResult;

/**

* ES操作 抽象方法 基本包含所有基本操作

*/

public interface ElasticSearchDao {

/**

* 刪除索引

* @param type ：當前刪除document名稱

* @return

*/

public JestResult deleteIndex(String type) ;

//清除緩存

public JestResult clearCache() ;

/**

* 關閉索引

* @param type ：文檔表示的對象類別

* @return

*/

public JestResult closeIndex(String type) ;

//優化索引

public JestResult optimizeIndex() ;

//刷新索引

public JestResult flushIndex();

//判斷索引是否存在

public JestResult indicesExists();

//查看節點信息

public JestResult nodesInfo();

//查看集群健康信息

public JestResult health();

//節點狀態

public JestResult nodesStats();

/**

* 更新Document

* @param index ：文檔在哪存放

* @param type ： 文檔表示的對象類別

* @param id ：文檔唯一標識

*/

public JestResult updateDocument(String script , String index,String type,String id);

/**

* 刪除Document

* @param index ：文檔在哪存放

* @param type ： 文檔表示的對象類別

* @param id ：文檔唯一標識

* @return

*/

public JestResult deleteDocument(String index,String type,String id) ;

/**

* 根據條件刪除

* @param index

* @param type

* @param params

*/

public JestResult deleteDocumentByQuery(String index, String type, String params);

/**

* 獲取Document

* @param o ：返回對象

* @param index ：文檔在哪存放

* @param type ： 文檔表示的對象類別

* @param id ：文檔唯一標識

* @return

*/

public <T> JestResult getDocument(T o , String index , String type , String id) ;

//Suggestion

public List<SuggestResult.Suggestion> suggest() ;

/**

* 查詢全部

* @param index ：文檔在哪存放

* @return

*/

public <T> List<Hit<T,Void>> searchAll(String index , T o);

/**

* 搜索

* @param keyWord ：搜索關鍵字

* @return

*/

public <T> List<Hit<T,Void>> createSearch(String keyWord , String type , T o , String... fields) ;

//bulkIndex操作

public <T> void bulkIndex(String index , String type , T o) ;

/**

* 創建索引

* @param o ：返回對象

* @param index ：文檔在哪存放

* @param type ： 文檔表示的對象類別

* @return

*/

public <T> JestResult createIndex(T o , String index , String type);

/**

* 搜索事件流圖表數據

* @param param

* @return

*/

public JsonObject searchEvent(String param);

}

package com.mdl.monitor.repositorys.elasticsearch.impl;

import java.io.IOException;

import java.lang.reflect.Method;

import java.util.Arrays;

import java.util.List;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.elasticsearch.index.query.QueryBuilders;

import org.elasticsearch.search.builder.SearchSourceBuilder;

import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Service;

import com.mdl.monitor.init.InitElasticSearchConfig;

import com.mdl.monitor.repositorys.elasticsearch.ElasticSearchDao;

import com.google.gson.JsonObject;

import com.google.gson.JsonParser;

import io.searchbox.client.JestResult;

import io.searchbox.cluster.Health;

import io.searchbox.cluster.NodesInfo;

import io.searchbox.cluster.NodesStats;

import io.searchbox.core.Bulk;

import io.searchbox.core.Delete;

import io.searchbox.core.Get;

import io.searchbox.core.Index;

import io.searchbox.core.Search;

import io.searchbox.core.SearchResult;

import io.searchbox.core.SearchResult.Hit;

import io.searchbox.core.Suggest;

import io.searchbox.core.SuggestResult;

import io.searchbox.core.SuggestResult.Suggestion;

import io.searchbox.core.Update;

import io.searchbox.indices.ClearCache;

import io.searchbox.indices.CloseIndex;

import io.searchbox.indices.DeleteIndex;

import io.searchbox.indices.Flush;

import io.searchbox.indices.IndicesExists;

import io.searchbox.indices.Optimize;

/**

* es操作實現類

*/

@Service

public class ElasticSearchDaoImpl implements ElasticSearchDao{

static protected final Log log = LogFactory.getLog(ElasticSearchDaoImpl.class.getName());

@Autowired

private InitElasticSearchConfig esConfig ;

@Override

public JestResult deleteIndex(String type) {

DeleteIndex deleteIndex = new DeleteIndex.Builder(type).build();

JestResult result = null ;

try {

result = esConfig.getClient().execute(deleteIndex);

log.info("deleteIndex == " + result.getJsonString());

} catch (IOException e) {

e.printStackTrace();

}

return result ;

}

@Override

public JestResult clearCache() {

ClearCache closeIndex = new ClearCache.Builder().build();

JestResult result = null ;

try {

result = esConfig.getClient().execute(closeIndex);

log.info("clearCache == " + result.getJsonString());

} catch (IOException e) {

e.printStackTrace();

}

return result ;

}

@Override

public JestResult closeIndex(String type) {

CloseIndex closeIndex = new CloseIndex.Builder(type).build();

JestResult result = null ;

try {

result = esConfig.getClient().execute(closeIndex);

log.info("closeIndex == " + result.getJsonString());

} catch (IOException e) {

e.printStackTrace();

}

return result ;

}

@Override

public JestResult optimizeIndex() {

Optimize optimize = new Optimize.Builder().build();

JestResult result = null ;

try {

result = esConfig.getClient().execute(optimize);

log.info("optimizeIndex == " + result.getJsonString());

} catch (IOException e) {

e.printStackTrace();

}

return result ;

}

@Override

public JestResult flushIndex() {

Flush flush = new Flush.Builder().build();

JestResult result = null ;

try {

result = esConfig.getClient().execute(flush);

log.info("flushIndex == " + result.getJsonString());

} catch (IOException e) {

e.printStackTrace();

}

return result ;

}

@Override

public JestResult indicesExists() {

IndicesExists indicesExists = new IndicesExists.Builder("article").build();

JestResult result = null ;

try {

result = esConfig.getClient().execute(indicesExists);

log.info("indicesExists == " + result.getJsonString());

} catch (IOException e) {

e.printStackTrace();

}

return result ;

}

@Override

public JestResult nodesInfo() {

NodesInfo nodesInfo = new NodesInfo.Builder().build();

JestResult result = null ;

try {

result = esConfig.getClient().execute(nodesInfo);

log.info("nodesInfo == " + result.getJsonString());

} catch (IOException e) {

e.printStackTrace();

}

return result ;

}

@Override

public JestResult health() {

Health health = new Health.Builder().build();

JestResult result = null ;

try {

result = esConfig.getClient().execute(health);

log.info("health == " + result.getJsonString());

} catch (IOException e) {

e.printStackTrace();

}

return result ;

}

@Override

public JestResult nodesStats() {

NodesStats nodesStats = new NodesStats.Builder().build();

JestResult result = null ;

try {

result = esConfig.getClient().execute(nodesStats);

log.info("nodesStats == " + result.getJsonString());

} catch (IOException e) {

e.printStackTrace();

}

return result ;

}

@Override

public JestResult updateDocument(String script , String index, String type, String id) {

/*String script = "{" +

" \"doc\" : {" +

" \"title\" : \""+article.getTitle()+"\"," +

" \"content\" : \""+article.getContent()+"\"," +

" \"author\" : \""+article.getAuthor()+"\"," +

" \"source\" : \""+article.getSource()+"\"," +

" \"url\" : \""+article.getUrl()+"\"," +

" \"pubdate\" : \""+new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss").format(article.getPubdate())+"\"" +

" }" +

"}";*/

Update update = new Update.Builder(script).index(index).type(type).id(id).build();

JestResult result = null ;

try {

result = esConfig.getClient().execute(update);

log.info("updateDocument == " + result.getJsonString());

} catch (IOException e) {

e.printStackTrace();

}

return result ;

}

@Override

public JestResult deleteDocument(String index, String type, String id) {

Delete delete = new Delete.Builder(id).index(index).type(type).build();

JestResult result = null ;

try {

result = esConfig.getClient().execute(delete);

log.info("deleteDocument == " + result.getJsonString());

} catch (IOException e) {

e.printStackTrace();

}

return result;

}

@Override

public JestResult deleteDocumentByQuery(String index, String type, String params) {

DeleteByQuery db = new DeleteByQuery.Builder(params)

.addIndex(index)

.addType(type)

.build();

JestResult result = null ;

try {

result = esConfig.getClient().execute(db);

log.info("deleteDocument == " + result.getJsonString());

} catch (IOException e) {

e.printStackTrace();

}

return result;

}

@Override

public <T> JestResult getDocument(T object , String index, String type, String id) {

Get get = new Get.Builder(index, id).type(type).build();

JestResult result = null ;

try {

result = esConfig.getClient().execute(get);

T o = (T) result.getSourceAsObject(object.getClass());

for (Method method : o.getClass().getMethods()) {

log.info("getDocument == " + method.getName());

}

} catch (IOException e) {

e.printStackTrace();

}

return result;

}

@Override

public List<Suggestion> suggest() {

String suggestionName = "my-suggestion";

Suggest suggest = new Suggest.Builder("{" +

" \"" + suggestionName + "\" : {" +

" \"text\" : \"the amsterdma meetpu\"," +

" \"term\" : {" +

" \"field\" : \"body\"" +

" }" +

" }" +

"}").build();

SuggestResult suggestResult = null ;

List<SuggestResult.Suggestion> suggestionList = null ;

try {

suggestResult = esConfig.getClient().execute(suggest);

log.info("suggestResult.isSucceeded() == " + suggestResult.isSucceeded());

suggestionList = suggestResult.getSuggestions(suggestionName);

log.info("suggestionList.size() == " + suggestionList.size());

for(SuggestResult.Suggestion suggestion:suggestionList){

System.out.println(suggestion.text);

}

} catch (IOException e) {

e.printStackTrace();

}

return suggestionList ;

}

@Override

public <T> List<Hit<T, Void>> searchAll(String index , T o) {

SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

searchSourceBuilder.query(QueryBuilders.matchAllQuery());

Search search = new Search.Builder(searchSourceBuilder.toString())

.addIndex(index)

.build();

SearchResult result = null ;

List<?> hits = null ;

try {

result = esConfig.getClient().execute(search);

System.out.println("本次查詢共查到："+result.getTotal()+"個關鍵字！");

log.info("本次查詢共查到："+result.getTotal()+"個關鍵字！");

hits = result.getHits(o.getClass());

} catch (IOException e) {

e.printStackTrace();

}

return (List<Hit<T, Void>>) hits ;

}

@Override

public <T> List<Hit<T, Void>> createSearch(String keyWord , String type , T o , String... fields) {

SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

searchSourceBuilder.query(QueryBuilders.queryStringQuery(keyWord));

HighlightBuilder highlightBuilder = new HighlightBuilder();

for(String field : fields){

highlightBuilder.field(field);//高亮field

}

highlightBuilder.preTags("<em>").postTags("</em>");//高亮標簽

highlightBuilder.fragmentSize(200);//高亮內容長度

searchSourceBuilder.highlighter(highlightBuilder);

Search search = new Search.Builder(searchSourceBuilder.toString()).addIndex(type).build();

SearchResult result = null ;

List<?> hits = null ;

try {

result = esConfig.getClient().execute(search);

System.out.println("本次查詢共查到："+result.getTotal()+"個結果！");

hits = result.getHits(o.getClass());

} catch (IOException e) {

e.printStackTrace();

}

return (List<Hit<T, Void>>) hits ;

}

@Override

public <T> void bulkIndex(String index, String type , T o) {

Bulk bulk = new Bulk.Builder()

.defaultIndex(index)

.defaultType(type)

.addAction(Arrays.asList(

new Index.Builder(o).build()

)).build();

try {

esConfig.getClient().execute(bulk);

} catch (IOException e) {

e.printStackTrace();

}

}

@Override

public <T> JestResult createIndex(T o, String index, String type) {

Index index1 = new Index.Builder(o).index(index).type(type).build();

JestResult jestResult = null ;

try {

jestResult = esConfig.getClient().execute(index1);

} catch (IOException e) {

e.printStackTrace();

}

return jestResult;

}

@Override

public JsonObject searchEvent(String param) {

JsonObject returnData = new JsonParser().parse(param).getAsJsonObject();

Search search = new Search.Builder(returnData.toString()).addType("event").addIndex("pi").build();

// Gson gs = new Gson();

// System.out.println("輸入參數為：" + "\n" + gs.toJson(search));

SearchResult result = null ;

try {

result = esConfig.getClient().execute(search);

// System.out.println("\n" + gs.toJson(result.getJsonObject()) + "\n" );

// System.out.println("本次查詢共查到：" + "\n" +result.getTotal()+"個結果！");

} catch (IOException e) {

e.printStackTrace();

}

return result.getJsonObject();

}

}

<!-- es連接配置 -->

<bean id="esConfig" class="com.mdl.monitor.init.InitElasticSearchConfig" >

<constructor-arg index="0" value="${elasticUrl}"/>

</bean>

@Override

public JsonObject searchEventHistogramByScroll(String scrollId) {

SearchScroll scroll = new SearchScroll.Builder(scrollId, "1m").build();

JestResult result = null ;

try {

result = esConfig.getClient().execute(scroll);

} catch (IOException e) {

e.printStackTrace();

}

return result.getJsonObject();

}

@Override

public JsonObject searchInitEventHistogram(String param) {

JsonObject returnData = new JsonParser().parse(param).getAsJsonObject();

Search search = new Search.Builder(returnData.toString())

.addIndex("pi")

.addType("event")

.setParameter(Parameters.SCROLL, "1m")

.build();

JestResult result = null;

try {

result = esConfig.getClient().execute(search);

} catch (IOException e) {

e.printStackTrace();

}

return result.getJsonObject();

}