package cn.test; import java.net.InetAddress; import java.net.UnknownHostException; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.action.search.SearchType; import org.elasticsearch.client.transport.TransportClient; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.transport.InetSocketTransportAddress; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.search.SearchHit; public class ElkTest { // private static final String CLUSTER_NAME = "cluster_name"; public static final String CLUSTER_NAME = "elasticsearch"; //實例名稱 private static final String IP = "127.0.0.1"; //private static final String IP = "192.168.0.29"; private static final int PORT = 9300; //端口 //1.設置集群名稱:默認是elasticsearch,並設置client.transport.sniff為true,使客戶端嗅探整個集群狀態,把集群中的其他機器IP加入到客戶端中 /* //對ES1.6有效 private static Settings settings = ImmutableSettings .settingsBuilder() .put("cluster.name",CLUSTER_NAME) .put("client.transport.sniff", true) .build(); */ //對ES2.0有效 private static Settings settings = Settings .settingsBuilder() .put("cluster.name",CLUSTER_NAME) .put("client.transport.sniff", true) .build(); //創建私有對象 private static TransportClient client; //反射機制創建單例的TransportClient對象 ES1.6版本 // static { // try { // Class<?> clazz = Class.forName(TransportClient.class.getName()); // Constructor<?> constructor = clazz.getDeclaredConstructor(Settings.class); // constructor.setAccessible(true); // client = (TransportClient) constructor.newInstance(settings); // client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(IP), PORT)); // } catch (Exception e) { // e.printStackTrace(); // } // } //ES2.0版本 static { try { client = TransportClient.builder().settings(settings).build() .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(IP), PORT)); } catch (UnknownHostException e) { e.printStackTrace(); } } //取得實例 public static synchronized TransportClient getTransportClient(){ return client; } //為集群添加新的節點 public static synchronized void addNode(String name){ try { client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(name),9300)); } catch (UnknownHostException e) { e.printStackTrace(); } } //刪除集群中的某個節點 public static synchronized void removeNode(String name){ try { client.removeTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(name),9300)); } catch (UnknownHostException e) { e.printStackTrace(); } } public static void main(String args[]){ String index="logstash-2016.02.16"; String type="logs"; SearchResponse response=ElkTest.getTransportClient().prepareSearch(index)//設置要查詢的索引(index) .setSearchType(SearchType.DFS_QUERY_THEN_FETCH) .setTypes(type)//設置type, 這個在建立索引的時候同時設置了, 或者可以使用head工具查看 .setQuery(QueryBuilders.matchQuery("message", "Accept")) //在這里"message"是要查詢的field,"Accept"是要查詢的內容 .setFrom(0) .setSize(10) .setExplain(true) .execute() .actionGet(); for(SearchHit hit:response.getHits()){ System.out.println(hit.getSourceAsString()); } } }