java調用Elasticserach的API實現日志查詢

前言：

    前面講了ELK+Filebeat搭建，點擊查看，那么如何在java程序中通過API查詢日志呢？我們知道，ELK的流程是FileBeat掃描日志文件，將日志發送到Logstash，Logstash將日志進行篩選和分組，發送給Elasticsearch，最終Kibnan通過調用Elasticsearch的接口，查詢日志，並展示出來。在這里，我的java程序就相當於替代了Kibana，由java調用es的接口，實現索引創建、索引查詢、索引刪除、日志數量查詢、日志列表查詢、日志分頁查詢等。

1，java項目引入Elasticsearch依賴

    這里使用的是maven，版本和es全家桶版本一致

<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch</artifactId>
    <version>7.14.0</version>
</dependency>
<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
    <version>7.14.0</version>
</dependency>

2，yml文件中加入es依賴

elasticsearch:
  host: 192.168.2.129
  port: 9200
  connTimeout: 3000
  socketTimeout: 5000
  connectionRequestTimeout: 500
  maxConnectNum: 100
  maxConnectPerRoute: 100

3，es的configuration文件

package com.test.elk;

import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 *
 * @author ：wdl
 * ==============================================================================
 * @文件名稱 ：ElasticsearchConfiguration.java
 * ==============================================================================
 * @本類描述 : ElsaticSearch配置類
 * @功能概述 ：
 * @創建日期 ：2021/10/12
 */
@Configuration
public class ElasticsearchConfiguration {
    @Value("${elasticsearch.host}")
    private String host;

    @Value("${elasticsearch.port}")
    private int port;

    @Value("${elasticsearch.connTimeout}")
    public int connectTimeOut;

    @Value("${elasticsearch.socketTimeout}")
    private int socketTimeOut;

    @Value("${elasticsearch.connectionRequestTimeout}")
    private int connectionRequestTimeOut;

    @Value("${elasticsearch.maxConnectNum}")
    private int maxConnectNum;

    @Value("${elasticsearch.maxConnectPerRoute}")
    private int maxConnectPerRoute;

    @Bean(destroyMethod = "close", name = "client")
    public RestHighLevelClient initRestClient() {
        // 異步httpclient連接延時配置
        RestClientBuilder builder = RestClient.builder(new HttpHost(host, port));
        builder.setRequestConfigCallback(builder1 -> {
                    builder1.setConnectTimeout(connectTimeOut);
                    builder1.setSocketTimeout(socketTimeOut);
                    builder1.setConnectionRequestTimeout(connectionRequestTimeOut);
                    return builder1;
                });
        // 異步httpclient連接數配置
        builder.setHttpClientConfigCallback(httpClientBuilder -> {
            httpClientBuilder.setMaxConnTotal(maxConnectNum);
            httpClientBuilder.setMaxConnPerRoute(maxConnectPerRoute);
            return httpClientBuilder;
        });
        return new RestHighLevelClient(builder);
    }
}

4，controller類

package com.test.elk;

import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.client.core.CountResponse;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;

import javax.annotation.Resource;
import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @author ：wdl
 * ==============================================================================
 * @文件名稱 ：ElasticSearchController.java
 * ==============================================================================
 * @本類描述 : 操作es控制類
 * @功能概述 ：
 * @創建日期 ：2021/10/12
 */
@Controller
@RequestMapping(value = "system/test")
public class ElasticSearchController {
    @Resource
    private RestHighLevelClient client;

    @RequestMapping(value = "existsIndex", method = RequestMethod.POST)
    @ResponseBody
    public boolean existsIndex(String index) throws IOException {
        // 聲明GetIndexRequest時要把索引信息帶進去，可以帶入字符串也可以帶入String[],如果傳入多個索引，則全部存在才會返回true,只要有一個不存在,就會返回false
        String[] indices = {index};
        GetIndexRequest request = new GetIndexRequest(indices);
        boolean exists = client.indices().exists(request, RequestOptions.DEFAULT);
        System.out.println("existsIndex: " + exists);
        return exists;
    }
    @RequestMapping(value = "createIndex", method = RequestMethod.POST)
    @ResponseBody
    public boolean createIndex (String index){
        try {
            if(!existsIndex(index)){
                CreateIndexRequest request = new CreateIndexRequest(index);
                this.client.indices().create(request, RequestOptions.DEFAULT);
                return Boolean.TRUE ;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return Boolean.FALSE;
    }
    @RequestMapping(value = "deleteIndex", method = RequestMethod.POST)
    @ResponseBody
    public boolean deleteIndex(String index) {
        try {
            if(existsIndex(index)){
                DeleteIndexRequest request = new DeleteIndexRequest(index);
                AcknowledgedResponse response = client.indices().delete(request, RequestOptions.DEFAULT);
                return response.isAcknowledged();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return Boolean.FALSE;
    }

    /**
     * 查詢總數
     */
    @RequestMapping(value = "count", method = RequestMethod.POST)
    @ResponseBody
    public Long count(String index){
        // 這里的條件和分頁查詢的條件一直，可以在分頁查詢中調用count方法，查詢出日志總數
        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
        // 匹配message字段中包含hello關鍵字的
        queryBuilder.must(QueryBuilders.matchPhraseQuery("message", "hello"));
        // 時間范圍篩選
        queryBuilder.filter(QueryBuilders.rangeQuery("@timestamp").from(1634086800000L).to(1634087400000L));
        CountRequest countRequest = new CountRequest(index);
        countRequest.query(queryBuilder);
        try {
            CountResponse countResponse = client.count(countRequest, RequestOptions.DEFAULT);
            return countResponse.getCount();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return 0L;
    }

    /**
     * 查詢集合
     */
    @RequestMapping(value = "list", method = RequestMethod.POST)
    @ResponseBody
    public List<Map<String,Object>> list (String index) {
        // RangeQueryBuilder 可以給字段指定時間范圍，但是不能加關鍵字查詢，如果需要多條件組合查詢，請看page方法
        RangeQueryBuilder rangequerybuilder = QueryBuilders
                .rangeQuery("@timestamp")
                .from(1634086800000L).to(1634087400000L);
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(rangequerybuilder);
        SearchRequest searchRequest = new SearchRequest(index);
        searchRequest.source(sourceBuilder);
        try {
            SearchResponse searchResp = client.search(searchRequest, RequestOptions.DEFAULT);
            List<Map<String,Object>> data = new ArrayList<>() ;
            SearchHit[] searchHitArr = searchResp.getHits().getHits();
            for (SearchHit searchHit:searchHitArr){
                Map<String,Object> res = new HashMap<>();
                Map<String,Object> temp = searchHit.getSourceAsMap();
                res.put("id",searchHit.getId());
                res.put("time",temp.get("@timestamp"));
                res.put("message",temp.get("message"));
                data.add(res);
            }
            return data;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 分頁查詢
     */
    @RequestMapping(value = "page", method = RequestMethod.POST)
    @ResponseBody
    public List<Map<String,Object>> page (String index) {
        // SearchSourceBuilder可以指定復雜條件，from：分頁查詢起始偏移量；size：每頁條數；sort：排序；query：查詢條件；可以聲明QueryBuilder的一個實現類，將條件加入queryBuilder，再放入sourceBuilder；
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.from(0);
        sourceBuilder.size(10);
        sourceBuilder.sort("@timestamp", SortOrder.DESC);
        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
        queryBuilder.must(QueryBuilders.matchPhraseQuery("message", "hello"));
        queryBuilder.filter(QueryBuilders.rangeQuery("@timestamp").from(1634086800000L).to(1634087400000L));
        sourceBuilder.query(queryBuilder);
        SearchRequest searchRequest = new SearchRequest(index);
        // 將sourceBulider放入searchRequest
        searchRequest.source(sourceBuilder);
        try {
            SearchResponse searchResp = client.search(searchRequest, RequestOptions.DEFAULT);
            List<Map<String,Object>> data = new ArrayList<>() ;
            SearchHit[] searchHitArr = searchResp.getHits().getHits();
            for (SearchHit searchHit:searchHitArr){
                Map<String,Object> res = new HashMap<>();
                Map<String,Object> temp = searchHit.getSourceAsMap();
                res.put("time",transTime(temp.get("@timestamp")));
                res.put("message",temp.get("message"));
                data.add(res);
            }
            return data;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null ;
    }
    //將Z時區時間轉換成UTC時間
    private Date transTime(Object time){
        Date date = new Date();
        String dateStr = String.valueOf(time);
        dateStr = dateStr.replace("Z", " UTC");
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS Z");
        try {
            date = format.parse(dateStr);
        } catch (ParseException e) {
            e.printStackTrace();
        }
        return date;
    }
}