【ES系列九】——批量同步数据至ES

package com.yunshi.timedtask.dao;
import org.apache.http.HttpHost;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@Component
public class TestElasticSearch4J {
  private static RestHighLevelClient client = new RestHighLevelClient(
          RestClient.builder(
                  new HttpHost("es1.yunshicloud.com", 9200, "http")
          ));
  public static void main(String[] args) throws IOException {
    TestElasticSearch4J testElasticSearch4J = new TestElasticSearch4J();
    List<Map<String,Object>> mapList = new ArrayList<>();
    System.out.println("准备数据，总计"+mapList.size()+"条");
    testElasticSearch4J.batchInsert(mapList,"material");
    client.close();
  }
  public  void batchInsert(List<Map<String,Object>> mapList,String type) throws IOException {
    BulkRequest request = new BulkRequest();
    for (Map<String,Object> map : mapList) {
      IndexRequest indexRequest= new IndexRequest("dev-rms-resource", type).source(map);
      request.add(indexRequest);
    }
    BulkResponse bulkItemResponses = client.bulk(request);
    for (BulkItemResponse bulkItemResponse : bulkItemResponses) {
      if (bulkItemResponse.isFailed()) {
        BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
        System.out.println("批量插入完成"+failure.getStatus());
      }
    }
  }
}

package com.yunshi.timedtask.scheduler;
import com.alibaba.fastjson.JSON;
import com.yunshi.timedtask.common.Constants;
import com.yunshi.timedtask.dao.BasicDao;
import com.yunshi.timedtask.dao.IEsBasicDao;
import com.yunshi.timedtask.dao.TestElasticSearch4J;
import com.yunshi.timedtask.dao.es.ESConfig;
import com.yunshi.timedtask.domain.*;
import com.yunshi.timedtask.util.DateUtil;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;
import java.io.*;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
 * Created by makang on 2019年12月30日.
 */
public class InitDataTaskService implements Runnable {
    private final Logger log = LoggerFactory.getLogger(InitDataTaskService.class);
    //调用底层的bean
    private BasicDao basicDao;
    private IEsBasicDao iEsBasicDao;
    private ESConfig esConfig;
    private int pageNum;
    //重写的构造器，用以接收外部传递过来的参数
    public InitDataTaskService(BasicDao basicDao, IEsBasicDao iEsBasicDao, ESConfig esConfig, int pageNum){
        this.basicDao = basicDao;
        this.iEsBasicDao = iEsBasicDao;
        this.esConfig = esConfig;
        this.pageNum = pageNum;
    }
    //实现的Runnable接口中的run方法
    @Override
    public  void run(){
        //开始时间
        long stertTime = System.currentTimeMillis();
        Map<String,Object> querymaterialAll = new HashMap<String,Object>(1);
        Map<String,Object> sortFilter = new HashMap<String,Object>(1);
        int pageNum = this.pageNum;
        //一、同步素材表中数据
        long materialAllSize = basicDao.count("material",querymaterialAll);
        log.info("开始同步素材，素材总数为："+materialAllSize);
        //1.1.添加检索条件来提高查询效率
        long allMaterialSize = 0;
        for (int pageSize=1;((pageSize-1)*pageNum*20)<materialAllSize;pageSize++){
            int materialSize = 0;
            log.info("当前执行分页条件为：pageSize:"+pageSize+"==pageNum:"+pageNum*20);
            List<Map<String,Object>> materialList = basicDao.find("material",sortFilter,querymaterialAll,pageSize,pageNum*20);
            try {
                String filePath = "./material"+pageSize+".json";
                for (Map<String,Object> materialMap:materialList) {
                    Map<String,Object> materialesMap = new HashMap<>();
                    materialesMap.putAll(materialMap);
                    //写入到json文件中
                    FileWriter fw = new FileWriter(filePath, true);
                    BufferedWriter bw = new BufferedWriter(fw);
                    bw.append("{\"index\":{\"_index\":\""+esConfig.getIndexname()+"\",\"_type\":\"material\"}} \n" );
                    bw.append(JSON.toJSONString(materialesMap)+" \n");
                    bw.close();
                    fw.close();
                    materialSize ++;
                    allMaterialSize++;
                }
                //执行提交命令
                execCommand("curl -XPOST  "+esConfig.getAnalyzerServerIp()+"/_bulk --data-binary @material"+pageSize+".json","./logs/material"+pageSize+".log");
            } catch (Exception e) {
                log.info("失败的素材页数为:"+pageSize);
                e.printStackTrace();
            }
            log.info("当前更新条数为：mongodb内容:"+materialSize);
            materialList.clear();
        }
        long endTime = System.currentTimeMillis();
        log.info("===素材更新用时为："+DateUtil.getDistanceTime(stertTime,endTime)
                +"===素材更新条数为："+allMaterialSize);
    }
    /**
     * 执行curl命令的方法
     * @param cmd
     * @param logPath
     */
    public void execCommand(String cmd,String logPath) {
        try {
            log.info("开始执行shell命令，同步es数据");
            long startTime = System.currentTimeMillis();
            Runtime rt = Runtime.getRuntime();
            Process proc = rt.exec(cmd,null,null);
            InputStream stderr = proc.getInputStream();
            InputStreamReader isr = new InputStreamReader(stderr, "GBK");
            BufferedReader br = new BufferedReader(isr);
            String line = "";
            int lineInt = 0;
            FileWriter fw = new FileWriter(logPath, true);
            BufferedWriter bw = new BufferedWriter(fw);
            while ((line = br.readLine()) != null) {
                lineInt++;
                bw.append(line+" \n");
            }
            bw.close();
            fw.close();
            long endTime = System.currentTimeMillis();
            log.info("结束执行shell命令，同步es数据，用时："+(endTime-startTime)+"==子进程执行个数："+lineInt);
            stderr.close();
            isr.close();
            br.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}