java多线程监控数据库表_干货分享：利用Java多线程技术导入数据到Elasticsearch

逃离我推掉我的手 2022-11-09 11:24 131阅读 0赞

前言

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

关键技术栈

Elasticsearch

jdbc

ExecutorService\\Thread

sql

工具说明

maven依赖

mysql

mysql-connector-java

$\{mysql.version\}

org.elasticsearch

elasticsearch

$\{elasticsearch.version\}

org.elasticsearch.client

transport

$\{elasticsearch.version\}

org.projectlombok

lombok

$\{lombok.version\}

com.alibaba

fastjson

$\{fastjson.version\}

java线程池设置

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

privatestaticintTHREADS =21;

publicstaticExecutorService POR = Executors.newFixedThreadPool(THREADS);

publicstaticExecutorService ROR = Executors.newFixedThreadPool(THREADS);

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

publicclassZlPendProducerimplementsRunnable \{

...

@Override

publicvoidrun() \{

System.out.println(threadName + "::启动...");

for(intj =0; j

try\{

....

intsize =1000;

for(inti =0; i

if(i + size > count) \{

//作用为size最后没有100条数据则剩余几条newList中就装几条

size = count - i;

String sql = "select \* from "+ tableName +" limit "+ i +", "+ size;

System.out.println(tableName + "::sql::"+ sql);

rs = statement.executeQuery(sql);

List lst = newArrayList<>();

while(rs.next()) \{

HistPendingEntity p = PendUtils.getHistPendingEntity(rs);

lst.add(p);

MteExecutor.POR.submit(newZlPendConsumer(lst));

Thread.sleep(2000);

....

\} catch(Exception e) \{

e.printStackTrace();

publicclassZlReadProducerimplementsRunnable \{

...已阅生产者处理逻辑同已办生产者

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

publicclassZlPendConsumerimplementsRunnable \{

privateString threadName;

privateList lst;

publicZlPendConsumer(List lst) \{

this.lst = lst;

@Override

publicvoidrun() \{

...

lst.forEach(v -> \{

try\{

String json = newGson().toJson(v);

EsClient.addDataInJSON(json, Const.ES.HistPendDB\_Index, Const.ES.HistPendDB\_type, v.getPendingId(), null);

Const.COUNTER.LD\_P.incrementAndGet();

\} catch(Exception e) \{

e.printStackTrace();

System.out.println("err::PendingId::"+ v.getPendingId());

\});

...

publicclassZlReadConsumerimplementsRunnable \{

//已阅消费者处理逻辑同已办消费者

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

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

publicvoidmonitorToES() \{

newThread(() -> \{

while(true) \{

StringBuilder sb = newStringBuilder();

sb.append("已办表数::").append(Const.TBL.TBL\_PEND\_COUNT)

.append("::已办总数::").append(Const.COUNTER.LD\_P\_TOTAL)

.append("::已办入库总数::").append(Const.COUNTER.LD\_P);

sb.append("~~~~已阅表数::").append(Const.TBL.TBL\_READ\_COUNT);

sb.append("::已阅总数::").append(Const.COUNTER.LD\_R\_TOTAL)

.append("::已阅入库总数::").append(Const.COUNTER.LD\_R);

if(ldPrevPendCount ==0&& ldPrevReadCount ==0) \{

ldPrevPendCount = Const.COUNTER.LD\_P.get();

ldPrevReadCount = Const.COUNTER.LD\_R.get();

start = System.currentTimeMillis();

\} else\{

longend = System.currentTimeMillis();

if((end - start) /1000>=60) \{

start = end;

sb.append("\\n\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\\n");

sb.append("已办每分钟TPS::"+ (Const.COUNTER.LD\_P.get() - ldPrevPendCount) +"条");

sb.append("::已阅每分钟TPS::"+ (Const.COUNTER.LD\_R.get() - ldPrevReadCount) +"条");

ldPrevPendCount = Const.COUNTER.LD\_P.get();

ldPrevReadCount = Const.COUNTER.LD\_R.get();

System.out.println(sb.toString());

try\{

Thread.sleep(3000);

\} catch(InterruptedException e) \{

e.printStackTrace();

\}).start();

初始化Elasticsearch：EsClient

String cName = meta.get("cName");//es集群名字

String esNodes = meta.get("esNodes");//es集群ip节点

Settings esSetting = Settings.builder()

.put("cluster.name", cName)

.put("client.transport.sniff",true)//增加嗅探机制，找到ES集群

.put("thread\_pool.search.size",5)//增加线程池个数，暂时设为5

.build();

String\[\] nodes = esNodes.split(",");

client = newPreBuiltTransportClient(esSetting);

for(String node : nodes) \{

if(node.length() >0) \{

String\[\] hostPort = node.split(":");

client.addTransportAddress(newTransportAddress(InetAddress.getByName(hostPort\[0\]), Integer.parseInt(hostPort\[1\])));

初始化数据库连接

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

启动参数

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

参数说明

ES-Cluster2019 为Elasticsearch集群名字

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

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

程序入口：MteMain

// 监控线程

Monitor monitorService = newMonitor();

monitorService.monitorToES();

// 已办生产者线程

Thread pendProducerThread = newThread(newZlPendProducer(conn,"ZlPendProducer"));

pendProducerThread.start();

// 已阅生产者线程

Thread readProducerThread = newThread(newZlReadProducer(conn,"ZlReadProducer"));

readProducerThread.start();

【编辑推荐】

java多线程监控数据库表_干货分享：利用Java多线程技术导入数据到Elasticsearch

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 java使用多线程插入数据到数据库

相关【Java-多线程】Java-多线程-Callable

相关多线程——基础多线程学习分享

相关 java多线程导入excel

相关 java多线程监控数据库表_干货分享：利用Java多线程技术导入数据到Elasticsearch

相关 java多线程_Java多线程编程核心技术（一）Java多线程技能（上）

相关 Java 多线程-多线程通信

相关多线程批处理数据导入工具 Java

相关 JAVA多线程（四）多线程数据共享

相关 Java多线程编程核心技术 —— Java多线程技能