135-1821-9792

干货分享：利用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.wtcwzsj.com/article/dposshi.html

其他资讯


关于我们
我们的服务
网站建设案例
新闻动态
联系我们

135-1821-9792

公司服务热线
友情链接： 成都集团网站建设   njwzjz.com   网站推广优化   都江堰发电机   成都发电机组维护   成都网站运营维护   景观雕塑制作   什邡网站建设   信衡房屋鉴定   合江网站建设   

Copyright © 2009-2022 www.wtcwzsj.com 青羊区广皓图文设计工作室（个体工商户） 版权所有　蜀ICP备19037934号