Java High Level REST Client API 로 작 성 된 ElasticSearch 도구 클래스

/**
 * elasticSearh    
 * @author Stephen
 * @version 1.0
 * @date 2018/09/17 11:12:08
 */
public class ElasticClient implements Closeable {
    
    private static final String INDEX_KEY = "index";
    private static final String TYPE_KEY = "type";   
    private static final String INDEX = "spider";
    private static final String TYPE = "doc";
    private static final String TIMESTAMP = "timestamp";
    
    private static final Logger LOGGER = LoggerFactory.getLogger(ElasticClient.class);
    
    private String[] hosts;
    
    protected RestHighLevelClient client;
    
    public ElasticClient(String[] hosts) {
        this.hosts = hosts;
    }

    @Override
    public void close() throws IOException {
        if (Objects.nonNull(client)) {
            client.close();
        }
    }
    
    /**
     *      
     */
    public void configure() {
        Validate.noNullElements(hosts, "Elastic        ！");
        HttpHost[] httpHosts = Arrays.stream(hosts).map(host -> {
            String[] hostParts = host.split(":");
            String hostName = hostParts[0];
            int port = Integer.parseInt(hostParts[1]);
            return new HttpHost(hostName, port, Const.SCHEMA);
        }).filter(Objects::nonNull).toArray(HttpHost[]::new);       
        client = new RestHighLevelClient(RestClient.builder(httpHosts));
    }
    
    /**
     *     (      5     1)
     * @param indexName
     * @throws IOException
     */
    public void createIndex(String indexName) throws IOException {
        if (checkIndexExists(indexName)) {
            LOGGER.error("\"index={}\"      ！", indexName);
            return;
        }
        CreateIndexRequest request = new CreateIndexRequest(indexName);
        request.mapping(TYPE, generateBuilder());        
        CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);
        //               
        boolean acknowledged = response.isAcknowledged();
        //                              
        boolean shardsAcknowledged = response.isShardsAcknowledged();
        if (acknowledged || shardsAcknowledged) {
            LOGGER.info("      ！     {}", indexName);
        }
    }
    
    /**
     *     (   index   type)
     * @param index
     * @param type
     * @throws IOException
     */
    public void createIndex(String index, String type) throws IOException {
        if (checkIndexExists(index)) {
            LOGGER.error("\"index={}\"     ！", index);
            return;
        }
        CreateIndexRequest request = new CreateIndexRequest(index);
        request.mapping(type, generateBuilder());        
        CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);
        boolean acknowledged = response.isAcknowledged();
        boolean shardsAcknowledged = response.isShardsAcknowledged();
        if (acknowledged || shardsAcknowledged) {
            LOGGER.info("      ！     {}", index);
        }
    }
    
    /**
     *     (    ：   、   )
     * @param indexName
     * @param shards
     * @param replicas
     * @throws IOException
     */
    public void createIndex(String indexName, int shards, int replicas) throws IOException {
        if (checkIndexExists(indexName)) {
            LOGGER.error("\"index={}\"     ！", indexName);
            return;
        }
        Builder builder = Settings.builder().put("index.number_of_shards", shards).put("index.number_of_replicas", replicas);
        CreateIndexRequest request = new CreateIndexRequest(indexName).settings(builder);
        request.mapping(TYPE, generateBuilder());
        CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);
        if (response.isAcknowledged() || response.isShardsAcknowledged()) {
            LOGGER.info("      ！     {}", indexName);
        }
    }
    
    /**
     *     
     * @param indexName
     * @throws IOException
     */
    public void deleteIndex(String indexName) throws IOException {
        try {
            DeleteIndexResponse response = client.indices().delete(new DeleteIndexRequest(indexName), RequestOptions.DEFAULT);
            if (response.isAcknowledged()) {
                LOGGER.info("{}       ！", indexName);
            }
        } catch (ElasticsearchException ex) {
            if (ex.status() == RestStatus.NOT_FOUND) {
                LOGGER.error("{}       ", indexName);
            }
            LOGGER.error("    ！");
        }
    }
    
    /**
     *         
     * @param indexName
     * @return
     * @throws IOException
     */
    public boolean checkIndexExists(String indexName) {
        GetIndexRequest request = new GetIndexRequest().indices(indexName);
        try {
            return client.indices().exists(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            LOGGER.error("        ，    ！");
        }
        return false;
    }
    
    /**
     *     
     * @param indexName
     * @throws IOException
     */
    public void openIndex(String indexName) throws IOException{
        if (!checkIndexExists(indexName)) {
            LOGGER.error("     ！");
            return;
        }
        OpenIndexRequest request = new OpenIndexRequest(indexName);
        OpenIndexResponse response = client.indices().open(request, RequestOptions.DEFAULT);
        if (response.isAcknowledged() || response.isShardsAcknowledged()) {
            LOGGER.info("{}       ！", indexName);
        }
    }
    
    /**
     *     
     * @param indexName
     * @throws IOException
     */
    public void closeIndex(String indexName) throws IOException {
        if (!checkIndexExists(indexName)) {
            LOGGER.error("     ！");
            return;
        }
        CloseIndexRequest request = new CloseIndexRequest(indexName); 
        CloseIndexResponse response = client.indices().close(request, RequestOptions.DEFAULT);
        if (response.isAcknowledged()) {
            LOGGER.info("{}      ！", indexName);
        }
    }
    
    /**
     *          (     message     ik   )
     * @param index
     * @param type
     * @throws IOException
     */
    public void setFieldsMapping(String index, String type) {
        PutMappingRequest request = new PutMappingRequest(index).type(type);
        try {
            request.source(generateBuilder());
            PutMappingResponse response = client.indices().putMapping(request, RequestOptions.DEFAULT);
            if (response.isAcknowledged()) {
                LOGGER.info("    \"index={}, type={}\"         ！", index, type);
            }
        } catch (IOException e) {
            LOGGER.error("\"index={}, type={}\"           ，     ！", index, type);
        }
    }
    private XContentBuilder generateBuilder() throws IOException {
        XContentBuilder builder = XContentFactory.jsonBuilder();        
        builder.startObject();
            builder.startObject("properties");           
                builder.startObject("message");               
                    builder.field("type", "text");
                    //  message  ，       ik_smart(    )
                    builder.field("analyzer", "ik_smart");                
                builder.endObject();
                builder.startObject(TIMESTAMP);               
                    builder.field("type", "date");
                    //                  long  
                    builder.field("format", "epoch_millis");                
                builder.endObject();
            builder.endObject();        
        builder.endObject();        
        return builder;
    }
   
    /**
     *     
     * @param indexName
     * @param typeName
     * @param id
     * @param jsonStr
     */
    public void addDocByJson(String indexName, String typeName, String id, String jsonString) throws IOException{
        if (!JsonValidator.validate(jsonString)) {
            LOGGER.error("   json   ，    ！");
            return;
        }
        if (!checkIndexExists(indexName)) {
            createIndex(indexName, typeName);
        }       
        IndexRequest request = new IndexRequest(indexName, typeName, id).source(jsonString, XContentType.JSON);
        // request opType   INDEX(    id    document，CREATE       )
        // request.opType(DocWriteRequest.OpType.CREATE)
        IndexResponse response = null;
        try {
            response = client.index(request, RequestOptions.DEFAULT);
            
            String index = response.getIndex();
            String type = response.getType();
            String documentId = response.getId();
            if (response.getResult() == DocWriteResponse.Result.CREATED) {
                LOGGER.info("      ！ index: {}, type: {}, id: {}", index , type, documentId);
            } else if (response.getResult() == DocWriteResponse.Result.UPDATED) {
                LOGGER.info("      ！ index: {}, type: {}, id: {}", index , type, documentId);
            }
            //       
            ReplicationResponse.ShardInfo shardInfo = response.getShardInfo();
            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
               LOGGER.error("           ！"); 
            }
            //          ，          
            if (shardInfo.getFailed() > 0) {
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    String reason = failure.reason(); 
                    LOGGER.error("      ：{}", reason);
                }
            }
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.CONFLICT) {
                LOGGER.error("    ！");
            }
            LOGGER.error("      ！");
        }
    }
    
    /**
     *     
     * @param index
     * @param type
     * @param id
     * @return
     * @throws IOException
     */
    public Map getDocument(String index, String type, String id) throws IOException{
        Map resultMap = new HashMap<>();        
        GetRequest request = new GetRequest(index, type, id);
        //   ( )
        request.realtime(false);
        //         ( )
        request.refresh(true);
        
        GetResponse response = null;
        try {
            response = client.get(request, RequestOptions.DEFAULT);
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.NOT_FOUND) {
                LOGGER.error("     ，     ！" );
            }
            if (e.status() == RestStatus.CONFLICT) {
                LOGGER.error("    ！" );
            }
            LOGGER.error("    ！");
        }
        
        if(Objects.nonNull(response)) {
            if (response.isExists()) { //           
                resultMap = response.getSourceAsMap();                               
            } else {
                //           。    ，         404    ，       GetResponse       。
                //               ，   isExists    false。
                LOGGER.error("     ，     ！" );
            }
        }
        return resultMap;
    }
    
    /**
     *     
     * @param index
     * @param type
     * @param id
     * @throws IOException
     */
    public void deleteDocument(String index, String type, String id) throws IOException {
        DeleteRequest request = new DeleteRequest(index, type, id);
        DeleteResponse response = null;
        try {
            response = client.delete(request, RequestOptions.DEFAULT);
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.CONFLICT) {
                LOGGER.error("    ！" );
            }
            LOGGER.error("    !");
        }
        if (Objects.nonNull(response)) {
            if (response.getResult() == DocWriteResponse.Result.NOT_FOUND) {
                LOGGER.error("      ，     ！");
            }
            LOGGER.info("     ！");
            ReplicationResponse.ShardInfo shardInfo = response.getShardInfo();
            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                LOGGER.error("         ");
            }
            if (shardInfo.getFailed() > 0) {
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    String reason = failure.reason();
                    LOGGER.error("    ：{}", reason);
                }
            }
        }
    }
    
    /**
     *         ( ："ctx._source.posttime=\"2018-09-18\"")    
     * @param index
     * @param type
     * @param id
     * @param script
     */
    public void updateDocByScript(String index, String type, String id, String script) throws IOException{
        Script inline = new Script(script);
        UpdateRequest request = new UpdateRequest(index, type, id).script(inline);
        try {
            UpdateResponse response  = client.update(request, RequestOptions.DEFAULT);            
            if (response.getResult() == DocWriteResponse.Result.UPDATED) {
                LOGGER.info("      ！");
            } else if (response.getResult() == DocWriteResponse.Result.DELETED) {
                LOGGER.error("\"index={},type={},id={}\"       ，    ！", response.getIndex(), response.getType(), response.getId());
            } else if(response.getResult() == DocWriteResponse.Result.NOOP) {
                LOGGER.error("       ！");
            }
            
            ReplicationResponse.ShardInfo shardInfo = response.getShardInfo();
            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                LOGGER.error("         ");
            }
            if (shardInfo.getFailed() > 0) {
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    String reason = failure.reason();
                    LOGGER.error("     ：{}", reason);
                }
            }
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.NOT_FOUND) {
                LOGGER.error("       ，     ！" );
            } else if (e.status() == RestStatus.CONFLICT) {
                LOGGER.error("      ！" );
            }
            LOGGER.error("    ！");
        }
    }
    
    /**
     *     JSON       (        ，           )
     * @param index
     * @param type
     * @param id
     * @param jsonString
     * @throws IOException
     */
    public void updateDocByJson(String index, String type, String id, String jsonString) throws IOException {
        if (!JsonValidator.validate(jsonString)) {
            LOGGER.error("   json   ，    ！");
            return;
        }
        if (!checkIndexExists(index)) {
            createIndex(index, type);
        }
        UpdateRequest request = new UpdateRequest(index, type, id);       
        request.doc(jsonString, XContentType.JSON);
        //            ，              
        request.docAsUpsert(true);
        try {            
            UpdateResponse response = client.update(request, RequestOptions.DEFAULT);
            String indexName = response.getIndex();
            String typeName = response.getType();
            String documentId = response.getId();
            if (response.getResult() == DocWriteResponse.Result.CREATED) {
                LOGGER.info("      ！index: {}, type: {}, id: {}", indexName, typeName, documentId);
            } else if (response.getResult() == DocWriteResponse.Result.UPDATED) {
                LOGGER.info("      ！");
            } else if (response.getResult() == DocWriteResponse.Result.DELETED) {
                LOGGER.error("\"index={},type={},id={}\"       ，    ！", indexName, typeName, documentId);
            } else if (response.getResult() == DocWriteResponse.Result.NOOP) {
                LOGGER.error("       ！");
            }
            
            ReplicationResponse.ShardInfo shardInfo = response.getShardInfo();
            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                LOGGER.error("         ");
            }
            if (shardInfo.getFailed() > 0) {
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    String reason = failure.reason();
                    LOGGER.error("     ：{}", reason);
                }
            }
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.NOT_FOUND) {
                LOGGER.error("       ，     ！" );
            } else if (e.status() == RestStatus.CONFLICT) {
                LOGGER.error("      ！" );
            }
            LOGGER.error("    ！");
        }
    }
    
    /**
     *       
     * @param params
     * @throws IOException
     */
    public void bulkAdd(List