干货分享：利用Java多线程技术导入数据到Elasticsearch

 前言

近期接到一个任务，需要改造现有从mysql往Elasticsearch导入数据MTE(mysqlToEs)小工具，由于之前采用单线程导入，千亿数据需要两周左右的时间才能导入完成，导入效率非常低。所以楼主花了3天的时间，利用java线程池框架Executors中的FixedThreadPool线程池重写了MTE导入工具，单台服务器导入效率提高十几倍（合理调整线程数据，效率更高）。

为鲁甸等地区用户提供了全套网页设计制作服务，及鲁甸网站建设行业解决方案。主营业务为网站设计制作、成都网站制作、鲁甸网站设计，以传统方式定制建设网站，并提供域名空间备案等一条龙服务，秉承以专业、用心的态度为用户提供真诚的服务。我们深信只要达到每一位用户的要求，就会得到认可，从而选择与我们长期合作。这样，我们也可以走得更远！

关键技术栈

  • Elasticsearch
  • jdbc
  • ExecutorService\Thread
  • sql

工具说明

maven依赖

 
 
 
    
  
  
  
  1.  
    2. 
  
  
  
  2.  mysql 
    3. 
  
  
  
  3.  mysql-connector-java 
    4. 
  
  
  
  4.  ${mysql.version} 
    5. 
  
  
  
  5.  
    6. 
  
  
  
  6.  
    7. 
  
  
  
  7.  org.elasticsearch 
    8. 
  
  
  
  8.  elasticsearch 
    9. 
  
  
  
  9.  ${elasticsearch.version} 
    10. 
  
  
  
  10.  
    11. 
  
  
  
  11.  
    12. 
  
  
  
  12.  org.elasticsearch.client 
    13. 
  
  
  
  13.  transport 
    14. 
  
  
  
  14.  ${elasticsearch.version} 
    15. 
  
  
  
  15.  
    16. 
  
  
  
  16.  
    17. 
  
  
  
  17.  org.projectlombok 
    18. 
  
  
  
  18.  lombok 
    19. 
  
  
  
  19.  ${lombok.version} 
    20. 
  
  
  
  20.  
    21. 
  
  
  
  21.  
    22. 
  
  
  
  22.  com.alibaba 
    23. 
  
  
  
  23.  fastjson 
    24. 
  
  
  
  24.  ${fastjson.version} 
    25. 
  
  
  
  25.  
    26.

java线程池设置

默认线程池大小为21个，可调整。其中POR为处理流程已办数据线程池，ROR为处理流程已阅数据线程池。

 
 
 
    
  
  
  
  1. private static int THREADS = 21; 
    2. 
  
  
  
  2. public static ExecutorService POR = Executors.newFixedThreadPool(THREADS); 
    3. 
  
  
  
  3. public static ExecutorService ROR = Executors.newFixedThreadPool(THREADS); 
    4.

定义已办生产者线程/已阅生产者线程：ZlPendProducer/ZlReadProducer

 
 
 
    
  
  
  
  1. public class ZlPendProducer implements Runnable { 
    2. 
  
  
  
  2.  ... 
    3. 
  
  
  
  3.  @Override 
    4. 
  
  
  
  4.  public void run() { 
    5. 
  
  
  
  5.  System.out.println(threadName + "::启动..."); 
    6. 
  
  
  
  6.  for (int j = 0; j < Const.TBL.TBL_PEND_COUNT; j++) 
    7. 
  
  
  
  7.  try { 
    8. 
  
  
  
  8.  .... 
    9. 
  
  
  
  9.  int size = 1000; 
    10. 
  
  
  
  10.  for (int i = 0; i < count; i += size) { 
    11. 
  
  
  
  11.  if (i + size > count) { 
    12. 
  
  
  
  12.  //作用为size***没有100条数据则剩余几条newList中就装几条 
    13. 
  
  
  
  13.  size = count - i; 
    14. 
  
  
  
  14.  } 
    15. 
  
  
  
  15.  String sql = "select * from " + tableName + " limit " + i + ", " + size; 
    16. 
  
  
  
  16.  System.out.println(tableName + "::sql::" + sql); 
    17. 
  
  
  
  17.  rs = statement.executeQuery(sql); 
    18. 
  
  
  
  18.  List lst = new ArrayList<>(); 
    19. 
  
  
  
  19.  while (rs.next()) { 
    20. 
  
  
  
  20.  HistPendingEntity p = PendUtils.getHistPendingEntity(rs); 
    21. 
  
  
  
  21.  lst.add(p); 
    22. 
  
  
  
  22.  } 
    23. 
  
  
  
  23.  MteExecutor.POR.submit(new ZlPendConsumer(lst)); 
    24. 
  
  
  
  24.  Thread.sleep(2000); 
    25. 
  
  
  
  25.  } 
    26. 
  
  
  
  26.  .... 
    27. 
  
  
  
  27.  } catch (Exception e) { 
    28. 
  
  
  
  28.  e.printStackTrace(); 
    29. 
  
  
  
  29.  } 
    30. 
  
  
  
  30.  } 
    31. 
  
  
  
  31. } 
    32. 
  
  
  
  32. public class ZlReadProducer implements Runnable { 
    33. 
  
  
  
  33.  ...已阅生产者处理逻辑同已办生产者 
    34. 
  
  
  
  34. } 
    35.

定义已办消费者线程/已阅生产者线程：ZlPendConsumer/ZlReadConsumer

 
 
 
    
  
  
  
  1. public class ZlPendConsumer implements Runnable { 
    2. 
  
  
  
  2.  private String threadName; 
    3. 
  
  
  
  3.  private List lst; 
    4. 
  
  
  
  4.  public ZlPendConsumer(List lst) { 
    5. 
  
  
  
  5.  this.lst = lst; 
    6. 
  
  
  
  6.  } 
    7. 
  
  
  
  7.  @Override 
    8. 
  
  
  
  8.  public void run() { 
    9. 
  
  
  
  9.  ... 
    10. 
  
  
  
  10.  lst.forEach(v -> { 
    11. 
  
  
  
  11.  try { 
    12. 
  
  
  
  12.  String json = new Gson().toJson(v); 
    13. 
  
  
  
  13.  EsClient.addDataInJSON(json, Const.ES.HistPendDB_Index, Const.ES.HistPendDB_type, v.getPendingId(), null); 
    14. 
  
  
  
  14.  Const.COUNTER.LD_P.incrementAndGet(); 
    15. 
  
  
  
  15.  } catch (Exception e) { 
    16. 
  
  
  
  16.  e.printStackTrace(); 
    17. 
  
  
  
  17.  System.out.println("err::PendingId::" + v.getPendingId()); 
    18. 
  
  
  
  18.  } 
    19. 
  
  
  
  19.  }); 
    20. 
  
  
  
  20.  ... 
    21. 
  
  
  
  21.  } 
    22. 
  
  
  
  22. } 
    23. 
  
  
  
  23. public class ZlReadConsumer implements Runnable { 
    24. 
  
  
  
  24.  //已阅消费者处理逻辑同已办消费者 
    25. 
  
  
  
  25. } 
    26.

定义导入Elasticsearch数据监控线程：Monitor

监控线程-Monitor为了计算每分钟导入Elasticsearch的数据总条数，利用监控线程，可以调整线程池的线程数的大小，以便利用多线程更快速的导入数据。

 
 
 
    
  
  
  
  1. public void monitorToES() { 
    2. 
  
  
  
  2.  new Thread(() -> { 
    3. 
  
  
  
  3.  while (true) { 
    4. 
  
  
  
  4.  StringBuilder sb = new StringBuilder(); 
    5. 
  
  
  
  5.  sb.append("已办表数::").append(Const.TBL.TBL_PEND_COUNT) 
    6. 
  
  
  
  6.  .append("::已办总数::").append(Const.COUNTER.LD_P_TOTAL) 
    7. 
  
  
  
  7.  .append("::已办入库总数::").append(Const.COUNTER.LD_P); 
    8. 
  
  
  
  8.  sb.append("~~~~已阅表数::").append(Const.TBL.TBL_READ_COUNT); 
    9. 
  
  
  
  9.  sb.append("::已阅总数::").append(Const.COUNTER.LD_R_TOTAL) 
    10. 
  
  
  
  10.  .append("::已阅入库总数::").append(Const.COUNTER.LD_R); 
    11. 
  
  
  
  11.  if (ldPrevPendCount == 0 && ldPrevReadCount == 0) { 
    12. 
  
  
  
  12.  ldPrevPendCount = Const.COUNTER.LD_P.get(); 
    13. 
  
  
  
  13.  ldPrevReadCount = Const.COUNTER.LD_R.get(); 
    14. 
  
  
  
  14.  start = System.currentTimeMillis(); 
    15. 
  
  
  
  15.  } else { 
    16. 
  
  
  
  16.  long end = System.currentTimeMillis(); 
    17. 
  
  
  
  17.  if ((end - start) / 1000 >= 60) { 
    18. 
  
  
  
  18.  start = end; 
    19. 
  
  
  
  19.  sb.append("\n#########################################\n"); 
    20. 
  
  
  
  20.  sb.append("已办每分钟TPS::" + (Const.COUNTER.LD_P.get() - ldPrevPendCount) + "条"); 
    21. 
  
  
  
  21.  sb.append("::已阅每分钟TPS::" + (Const.COUNTER.LD_R.get() - ldPrevReadCount) + "条"); 
    22. 
  
  
  
  22.  ldPrevPendCount = Const.COUNTER.LD_P.get(); 
    23. 
  
  
  
  23.  ldPrevReadCount = Const.COUNTER.LD_R.get(); 
    24. 
  
  
  
  24.  } 
    25. 
  
  
  
  25.  } 
    26. 
  
  
  
  26.  System.out.println(sb.toString()); 
    27. 
  
  
  
  27.  try { 
    28. 
  
  
  
  28.  Thread.sleep(3000); 
    29. 
  
  
  
  29.  } catch (InterruptedException e) { 
    30. 
  
  
  
  30.  e.printStackTrace(); 
    31. 
  
  
  
  31.  } 
    32. 
  
  
  
  32.  } 
    33. 
  
  
  
  33.  }).start(); 
    34. 
  
  
  
  34. } 
    35.

初始化Elasticsearch：EsClient

 
 
 
    
  
  
  
  1. String cName = meta.get("cName");//es集群名字 
    2. 
  
  
  
  2. String esNodes = meta.get("esNodes");//es集群ip节点 
    3. 
  
  
  
  3. Settings esSetting = Settings.builder() 
    4. 
  
  
  
  4.  .put("cluster.name", cName) 
    5. 
  
  
  
  5.  .put("client.transport.sniff", true)//增加嗅探机制，找到ES集群 
    6. 
  
  
  
  6.  .put("thread_pool.search.size", 5)//增加线程池个数，暂时设为5 
    7. 
  
  
  
  7.  .build(); 
    8. 
  
  
  
  8. String[] nodes = esNodes.split(","); 
    9. 
  
  
  
  9. client = new PreBuiltTransportClient(esSetting); 
    10. 
  
  
  
  10. for (String node : nodes) { 
    11. 
  
  
  
  11.  if (node.length() > 0) { 
    12. 
  
  
  
  12.  String[] hostPort = node.split(":"); 
    13. 
  
  
  
  13.  client.addTransportAddress(new TransportAddress(InetAddress.getByName(hostPort[0]), Integer.parseInt(hostPort[1]))); 
    14. 
  
  
  
  14.  } 
    15. 
  
  
  
  15. } 
    16.

初始化数据库连接

 
 
 
    
  
  
  
  1. conn = DriverManager.getConnection(url, user, password); 
    2.

启动参数

 
 
 
    
  
  
  
  1. nohup java -jar mte.jar ES-Cluster2019 node1:9300,node2:9300,node3:9300 root 123456! jdbc:mysql://ip:3306/mte 130 130 >> ./mte.log 2>&1 & 
    2.

参数说明

ES-Cluster2019 为Elasticsearch集群名字

node1:9300,node2:9300,node3:9300为es的节点IP

130 130为已办已阅分表的数据

程序入口：MteMain

    

 
 
 
    
  
  
  
  1. // 监控线程 
    2. 
  
  
  
  2. Monitor monitorService = new Monitor(); 
    3. 
  
  
  
  3. monitorService.monitorToES(); 
    4. 
  
  
  
  4. // 已办生产者线程 
    5. 
  
  
  
  5. Thread pendProducerThread = new Thread(new ZlPendProducer(conn, "ZlPendProducer")); 
    6. 
  
  
  
  6. pendProducerThread.start(); 
    7. 
  
  
  
  7. // 已阅生产者线程 
    8. 
  
  
  
  8. Thread readProducerThread = new Thread(new ZlReadProducer(conn, "ZlReadProducer")); 
    9. 
  
  
  
  9. readProducerThread.start(); 
    10.

网页名称：干货分享：利用Java多线程技术导入数据到Elasticsearch
URL分享：http://www.mswzjz.cn/qtweb/news34/403784.html

攀枝花网站建设、攀枝花网站运维推广公司-贝锐智能，是专注品牌与效果的网络营销公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 贝锐智能

贝锐智能技术为您推荐以下文章

App设计知识

分类信息网站

彭山网站制作　　 　成都雪糕加盟　　 　小谭建站　　 　珉田数据中心　　 　主动防护网　　 　四川主机托管　　 　重庆网站设计　　 　IDC机房托管　　 　成都门户网站建设　　 　成都网站推广　　 　重庆企业网站建设　　 　企业网站改版　　 　商标注册　　 　川西大数据中心　　 　成都工装公司　　 　公司网站建设　　 　成都商城网站制作　　 　中国电信成都枢纽中心　　 　茶叶批发零售　　 　网络整合营销