本篇文章给大家带来的内容是关于Java多线程批量数据导入的方法介绍，有一定的参考价值，有需要的朋友可以参考一下，希望对你有所帮助。
前言：当遇到大量数据导入时，为了提高处理的速度，可以选择使用多线程来批量处理这些处理。常见的场景有：
大文件导入数据库（这个文件不一定是标准的
可导入文件或者需要在内存中经过一定的处理）
数据同步（从第三方接口拉取数据处理后写入自己的数据库）
以上的场景有一个共性，这类数据导入的场景简单来说就是将
数据从一个数据源移动到另外一个数据源
，而其中必定可以分为两步
数据读取
：从数据源读取数据到内存
数据写入
：将内存中的数据写入到另外一个数据源，可能存在数据处理
而且数据读取的速度一般会比数据写入的速度快很多，即
读取快，写入慢
。
设计思路
由于场景的特点是
读取快，写入慢
，如果是使用多线程处理，建议是
数据写入
部分改造为多线程。而
数据读取
可以改造成批量读取数据。简单来说就是两个要点：
批量读取数据
多线程写入数据
示例
多线程批量处理最简单的方案是使用线程池来进行处理，下面会通过一个模拟批量读取和写入的服务，以及对这个服务的多线程写入调用作为示例，展示如何多线程批量数据导入。
模拟服务
批量数据处理器
测试类
运行结果
分析
在单线程情况下的执行时间应该为
，即
，而改造为多线程后执行时间为
。
Eclipse导入Android或其他的JAVA项目的正确方法 WORD版
本文档主要讲述的是Eclipse导入Android或其他的JAVA项目的正确方法；希望本文档会给有需要的朋友带来帮助；感兴趣的朋友可以过来看看
下载
立即学习
“
Java免费学习笔记（深入）
”；
【相关推荐：
Java视频教程
】
CSVimport java.util.concurrent.atomic.AtomicLong;
/**
* 数据批量写入用的模拟服务
*
* @author RJH
* create at 2019-04-01
*/
public class MockService {
/**
* 可读取总数
*/
private long canReadTotal;
/**
* 写入总数
*/
private AtomicLong writeTotal=new AtomicLong(0);
/**
* 写入休眠时间（单位：毫秒）
*/
private final long sleepTime;
/**
* 构造方法
*
* @param canReadTotal
* @param sleepTime
*/
public MockService(long canReadTotal, long sleepTime) {
this.canReadTotal = canReadTotal;
this.sleepTime = sleepTime;
}
/**
* 批量读取数据接口
*
* @param num
* @return
*/
public synchronized long readData(int num) {
long readNum;
if (canReadTotal >= num) {
canReadTotal -= num;
readNum = num;
} else {
readNum = canReadTotal;
canReadTotal = 0;
}
//System.out.println("read data size:" + readNum);
return readNum;
}
/**
* 写入数据接口
*/
public void writeData() {
try {
// 休眠一定时间模拟写入速度慢
Thread.sleep(sleepTime);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 写入总数自增
System.out.println("thread:" + Thread.currentThread() + " write data:" + writeTotal.incrementAndGet());
}
/**
* 获取写入的总数
*
* @return
*/
public long getWriteTotal() {
return writeTotal.get();
}
}import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* 基于线程池的多线程批量写入处理器
* @author RJH
* create at 2019-04-01
*/
public class SimpleBatchHandler {
private ExecutorService executorService;
private MockService service;
/**
* 每次批量读取的数据量
*/
private int batch;
/**
* 线程个数
*/
private int threadNum;
public SimpleBatchHandler(MockService service, int batch,int threadNum) {
this.service = service;
this.batch = batch;
//使用固定数目的线程池
this.executorService = Executors.newFixedThreadPool(threadNum);
}
/**
* 开始处理
*/
public void startHandle() {
// 开始处理的时间
long startTime = System.currentTimeMillis();
System.out.println("start handle time:" + startTime);
long readData;
while ((readData = service.readData(batch)) != 0) {// 批量读取数据，知道读取不到数据才停止
for (long i = 0; i < readData; i++) {
executorService.execute(() -> service.writeData());
}
}
// 关闭线程池
executorService.shutdown();
while (!executorService.isTerminated()) {//等待线程池中的线程执行完
}
// 结束时间
long endTime = System.currentTimeMillis();
System.out.println("end handle time:" + endTime);
// 总耗时
System.out.println("total handle time:" + (endTime - startTime) + "ms");
// 写入总数
System.out.println("total write num:" + service.getWriteTotal());
}
}/**
* SimpleBatchHandler的测试类
* @author RJH
* create at 2019-04-01
*/
public class SimpleBatchHandlerTest {
public static void main(String[] args) {
// 总数
long total=100000;
// 休眠时间
long sleepTime=100;
// 每次拉取的数量
int batch=100;
// 线程个数
int threadNum=16;
MockService mockService=new MockService(total,sleepTime);
SimpleBatchHandler handler=new SimpleBatchHandler(mockService,batch,threadNum);
handler.startHandle();
}
}start handle time:1554298681755
thread:Thread[pool-1-thread-2,5,main] write data:1
thread:Thread[pool-1-thread-1,5,main] write data:2
...省略部分输出
thread:Thread[pool-1-thread-4,5,main] write data:100000
end handle time:1554299330202
total handle time:648447ms
total write num:100000total*sleepTime10000000ms648447ms