Java 使用 HttpClient 調用 es restful api 操作es

前言

今天交付 ES 管理平台，因為 ES 有兩套集群，分別是5.x 和 6.x 為了代碼的通用性，需要把 Transport Client 的相關操作全部廢棄，改為直接調用 rest api

准備工作

        <dependency>
            <groupId>org.apache.httpcomponents</groupId>
            <artifactId>httpclient</artifactId>
            <version>4.5.3</version>
        </dependency>

        <dependency>
            <groupId>com.arronlong</groupId>
            <artifactId>httpclientutil</artifactId>
            <version>1.0.4</version>
        </dependency>
<!--        工具集-->
        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.3.8</version>
        </dependency>

相關操作

1.根據ip地址獲取所有索引

/**
     * 獲取所有索引index
     * @param ip
     * @return
     */
    public static List<IndexInfo> getAllIndex(String ip) {
        HttpConfig config = HttpConfig.custom().url("http://"+ip+":9200/_cat/indices?format=json&pretty");
        String jsonStr = null;
        try {
            jsonStr = HttpClientUtil.get(config);
        } catch (HttpProcessException e) {
            e.printStackTrace();
            return null;
        }
        JSONArray jsonArray = JSONUtil.parseArray(jsonStr);

        List<IndexInfo> list = new ArrayList<>();
        List<Map> maps = jsonArray.toList(Map.class);
        for(Map map : maps) {
            IndexInfo index = new IndexInfo();
            index.setHealth ((String)map.get("health"));
            index.setStatus ((String)map.get("status"));
            index.setIndex ((String)map.get("index"));
            index.setUuid ((String)map.get("uuid"));
            index.setPri ((String)map.get("pri"));
            index.setRep ((String)map.get("rep"));
            index.setDocs ((String)map.get("docs.count"));
            index.setDeleted ((String)map.get("docs.deleted"));
            index.setStoreSize ((String)map.get("store.size"));
            index.setPriStoreSize ((String)map.get("pri.store.size"));
            list.add(index);
        }
        return list;
    }

　　查詢結果解釋

• 索引健康（health），green為正常，yellow表示索引不可靠（單節點），red索引不可用。與集群健康狀態一致。
• 狀態（status），表明索引是否打開。
• 索引名稱（index），這里有.kibana和school。
• uuid，索引內部隨機分配的名稱，表示唯一標識這個索引。
• 主分片（pri），這個就是集群的主分片數。
• 文檔數（docs.count）。
• 已刪除文檔數（docs.deleted），這里統計了被刪除文檔的數量。
• 索引存儲的總容量（store.size），索引的總容量，是主分片總容量的兩倍，因為存在一個副本。
• 主分片的總容量（pri.store.size），主分片容量，是索引總容量的一半。

2.查看集群的健康狀態

/**
     * 查看集群的健康狀態
     *
     * @param ip
     * @return
     */
    public static List<Cluster> heathInfo(String ip) {
        HttpConfig config = HttpConfig.custom().url("http://" + ip + ":9200/_cat/health?format=json&pretty");
        String jsonStr = null;
        try {
            jsonStr = HttpClientUtil.get(config);
        } catch (HttpProcessException e) {
            e.printStackTrace();
            return null;
        }
        JSONArray jsonArray = JSONUtil.parseArray(jsonStr);

        List<Cluster> list = new ArrayList<>();
        List<Map> maps = jsonArray.toList(Map.class);
        for (Map map : maps) {
            Cluster cluster = new Cluster();
            cluster.setCluster((String) map.get("cluster"));
            cluster.setStatus((String) map.get("status"));
            cluster.setNodeTotal((String) map.get("node.total"));
            cluster.setNodeData((String) map.get("node.data"));
            cluster.setShards((String) map.get("shards"));
            cluster.setPri((String) map.get("pri"));
            cluster.setRelo((String) map.get("relo"));
            cluster.setInit((String) map.get("init"));
            cluster.setUnassign((String) map.get("unassign"));
            cluster.setPendingTasks((String) map.get("pending_tasks"));
            cluster.setMaxTaskWaitTime((String) map.get("max_task_wait_time"));
            cluster.setActiveShardsPercent((String) map.get("active_shards_percent"));
            list.add(cluster);
        }
        return list;
    }

• cluster ，集群名稱
• status，集群狀態 green代表健康；yellow代表分配了所有主分片，但至少缺少一個副本，此時集群數據仍舊完整；red代表部分主分片不可用，可能已經丟失數據。
• node.total，代表在線的節點總數量
• node.data，代表在線的數據節點的數量
• shards， active_shards 存活的分片數量
• pri，active_primary_shards 存活的主分片數量 正常情況下 shards的數量是pri的兩倍。
• relo， relocating_shards 遷移中的分片數量，正常情況為 0
• init， initializing_shards 初始化中的分片數量 正常情況為 0
• unassign， unassigned_shards 未分配的分片 正常情況為 0
• pending_tasks，准備中的任務，任務指遷移分片等 正常情況為 0
• max_task_wait_time，任務最長等待時間
• active_shards_percent，正常分片百分比 正常情況為 100%

3.獲取索引的基本信息（大小，文檔數，字段。。）

/**
     * 獲取索引基本信息
     */
    public static JSONObject getIndexInfo(String ip, String index) {
        // 獲取索引的別名，字段，創建時間
        HttpConfig config = HttpConfig.custom().url("http://" + ip + ":9200/" + index);
        String jsonStr = null;
        try {
            jsonStr = HttpClientUtil.get(config);
        } catch (HttpProcessException e) {
            e.printStackTrace();
            return null;
        }
        JSONObject jsonObject = JSONUtil.parseObj(jsonStr);
        JSONObject fields = jsonObject.getJSONObject(index).getJSONObject("mappings").getJSONObject("CrmLog").getJSONObject("properties");
        JSONObject aliases = jsonObject.getJSONObject(index).getJSONObject("aliases");
        JSONObject jsonIndex = jsonObject.getJSONObject(index).getJSONObject("settings").getJSONObject("index");
        // 時間戳
        String dateStr = jsonIndex.getStr("creation_date");
        Date date = new Date(Long.parseLong(dateStr));
        // 分片數
        String shards = (String) jsonIndex.get("number_of_shards");
        // 主分片數
        String pri = (String) jsonIndex.get("number_of_replicas");

        // 獲取索引所占內存大小
        config = HttpConfig.custom().url("http://" + ip + ":9200/" + index + "/_stats");
        try {
            jsonStr = HttpClientUtil.get(config);
        } catch (HttpProcessException e) {
            e.printStackTrace();
            return null;
        }
        jsonObject = JSONUtil.parseObj(jsonStr);
        Integer byteNum = (Integer) jsonObject.getJSONObject("indices").getJSONObject(index).getJSONObject("primaries").getJSONObject("store").get("size_in_bytes");
        Integer docs = (Integer) jsonObject.getJSONObject("indices").getJSONObject(index).getJSONObject("primaries").getJSONObject("docs").get("count");

        // 整合輸出結果
        String size;
        BigDecimal bigDecimal = new BigDecimal(byteNum);
        BigDecimal mb = bigDecimal.divide(BigDecimal.valueOf(1024 * 1024));
        BigDecimal gb = mb.divide(BigDecimal.valueOf(1024));
        if (gb.intValue() / 10 > 0) {
            size = gb.setScale(2, RoundingMode.HALF_UP) + " gb";
        } else {
            size = mb.setScale(2, RoundingMode.HALF_UP) + " mb";
        }

        JSONObject result = new JSONObject();
        list = getFields(fields);
        String[] aliaseArr = getAliases(aliases);
        result.putOpt("alise", aliaseArr);
        result.putOpt("date", DateUtil.format(date, "yyyy-MM-dd HH:mm:ss"));
        result.putOpt("shards", shards);
        result.putOpt("pri", pri);
        result.putOpt("size", size);
        result.putOpt("docs", docs);
        return result;
    }

    /**
     * 解析別名
     * @param aliases
     * @return
     */
    private static String[] getAliases(JSONObject aliases) {
        String[] aliaseArr = {};
        if(aliases.size() > 0) {
            aliaseArr = aliases.toString()
                    .replace("{", "")
                    .replace("}", "")
                    .replace(",", "")
                    .replace("\"", "")
                    .split(":");
        }
        return aliaseArr;
    }
/**
     * 解析出字段list
     * @param jsonObject
     * @return
     */
    private static List<ColumnModel> getFields(JSONObject jsonObject) {
        Map map = JSONUtil.toBean(jsonObject, Map.class);
        Set set = map.keySet();
        List<ColumnModel> list = new ArrayList<>();
        for(Object field : set) {
            String fname = (String) field;
            String type = jsonObject.getJSONObject(fname).getStr("type");
            ColumnModel columnModel = new ColumnModel();
            columnModel.setColumnName(fname);
            columnModel.setTypeName(type);
            list.add(columnModel);
        }
        return list;
    }

其他操作同理，只需要調用對應的 restful api 即可，拼接字符串過程比較冗長，在此不做展示了。有疑問可以留言