public static void init(String filepath) throws IOException, InterruptedException {  
        // https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/java-docs-bulk-processor.html  
        /*cluster.name 必须与服务端相同*/  
        Settings settings = Settings.builder().put("cluster.name", "my-application").build();  
//        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();  
//        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("elastic", "12346"));  //es账号密码（默认用户名为elastic）  
        TransportClient client = new PreBuiltTransportClient(settings)  
                .addTransportAddress(new TransportAddress(InetAddress.getByName(properties.getProperty("ip")), Integer.valueOf(properties.getProperty("port"))));  
        BulkProcessor build = BulkProcessor.builder(client, new BulkProcessor.Listener() {  
            @Override  
            public void beforeBulk(long l, BulkRequest bulkRequest) {  
                /*在批量执行之前调用此方法*/  
                logger.info("---尝试插入{}条数据---", bulkRequest.numberOfActions());  
            }  
            @Override  
            public void afterBulk(long l, BulkRequest bulkRequest, BulkResponse bulkResponse) {  
                /*批量执行后调用此方法。例如，您可以检查是否有一些失败的请求response.hasFailures()*/  
                logger.info("---插入完成-> {}条数据,---", bulkRequest.numberOfActions());  
                bulkResponse.hasFailures();  
            }  
            @Override  
            public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) {  
                /*批量失败并引发一个 Throwable*/  
                logger.error("error：" + throwable.getMessage() + " \ncause：" + throwable.getCause());  
//                logger.info("---尝试插入{}条数据---", bulkRequest.numberOfActions());  
            }  
        })  
                /*每x个请求执行一次Bulk*/  
                .setBulkActions(Integer.valueOf(properties.getProperty("setBulkActions")))  
                /*每5MB Bulk一次*/  
                .setBulkSize(new ByteSizeValue(Integer.valueOf(properties.getProperty("setBulkSize")), ByteSizeUnit.MB))  
                /*无论请求多少，我都希望每5秒刷新一次*/  
                .setFlushInterval(TimeValue.timeValueSeconds(Integer.valueOf(properties.getProperty("setFlushInterval"))))  
                /*设置并发请求数。值为0表示仅允许执行一个请求。值为1表示允许在累积新的批量请求时执行1个并发请求。*/  
                .setConcurrentRequests(Integer.valueOf(properties.getProperty("setConcurrentRequests")))  
                /*设置一个自定义退避策略，该策略最初将等待100毫秒，然后呈指数增长，然后重试最多3次。每当一个或多个批量项目请求失败时，都会尝试重试，EsRejectedExecutionException 并显示，表明有太多的计算资源可用于处理请求。要禁用退避，请通过BackoffPolicy.noBackoff()*/  
                .setBackoffPolicy(  
                        BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3)  
                ).build();  
                /*默认情况下BulkProcessor：  
                    将bulkActions设置为 1000  
                    设置bulkSize为 5mb  
                    没有设置flushInterval  
                    将parallelRequests设置为1，这意味着异步执行刷新操作。  
                    将backoffPolicy设置为具有8次重试和50ms启动延迟的指数补偿。总等待时间约为5.1秒。  
                */  
        String index = properties.getProperty("index");  
        logger.info("开始添加数据!");  
        Gson gson = new Gson();  
        // 读取日志文件  
        File file = new File(filepath);  
        BufferedInputStream fis = new BufferedInputStream(new FileInputStream(file));  
        BufferedReader reader = new BufferedReader(new InputStreamReader(fis, "utf-8"), 10 * 1024 * 1024);// 用10M的缓冲读取文本文件  
        String line = "";  
        while ((line = reader.readLine()) != null) {  
            //TODO: write your business  
            Map<String, String> map = new HashMap<>();  
            /*第一种情况*/  
            String[] split = line.split("----");  
            if (split.length > 1) {  
                map.put("qq", split[1]);  
                map.put("phone", split[2]);  
            }  
            if (map.size() > 0) {  
                /*添加数据*/  
                build.add(new IndexRequest(index, "_doc").source(gson.toJson(map), XContentType.JSON));  
            }  
        }  
        logger.info("添加数据完成!等待提交......");  
        /*刷新剩余请求*/  
        //build.flush();  
        build.awaitClose(10, TimeUnit.MINUTES);  
//        build.close();