阅读文本大概需要3分钟。
来源&＃xff1a;https://www.cnblogs.com/barrywxx/p/10700221.html
最近业务方有一个需求&＃xff0c;需要一次导入超过100万数据到系统数据库。可能大家首先会想&＃xff0c;这么大的数据&＃xff0c;干嘛通过程序去实现导入&＃xff0c;为什么不直接通过SQL导入到数据库。
一、为什么一定要在代码实现
说说为什么不能通过SQL直接导入到数据库&＃xff0c;而是通过程序实现&＃xff1a;
1. 首先&＃xff0c;这个导入功能开始提供页面导入&＃xff0c;只是开始业务方保证的一次只有<3W的数据导入&＃xff1b;
2. 其次&＃xff0c;业务方导入的内容需要做校验&＃xff0c;比如门店号&＃xff0c;商品号等是否系统存在&＃xff0c;需要程序校验&＃xff1b;
3. 最后&＃xff0c;业务方导入的都是编码&＃xff0c;数据库中还要存入对应名称&＃xff0c;方便后期查询&＃xff0c;SQL导入也是无法实现的。
基于以上上三点&＃xff0c;就无法直接通过SQL语句导入数据库。那就只能老老实实的想办法通过程序实现。
二、程序实现有以下技术难点
1. 一次读取这么大的数据量&＃xff0c;肯定会导致服务器内存溢出&＃xff1b;
2. 调用接口保存一次传输数据量太大&＃xff0c;网络传输压力会很大&＃xff1b;
3. 最终通过SQL一次批量插入&＃xff0c;对数据库压力也比较大&＃xff0c;如果业务同时操作这个表数据&＃xff0c;很容易造成死锁。
三、解决思路
根据列举的技术难点我的解决思路是&＃xff1a;
1. 既然一次读取整个导入文件&＃xff0c;那就先将文件流上传到服务器磁盘&＃xff0c;然后分批从磁盘读取(支持多线程读取)&＃xff0c;这样就防止内存溢出&＃xff1b;
2. 调用插入数据库接口也是根据分批读取的内容进行调用&＃xff1b;
3. 分批插入数据到数据库。
四、具体实现代码
1. 流式上传文件到服务器磁盘
略&＃xff0c;一般Java上传就可以实现&＃xff0c;这里就不贴出。
2. 多线程分批从磁盘读取
批量读取文件&＃xff1a;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
/**
* 类功能描述&＃xff1a;批量读取文件
*
* &＃64;author WangXueXing create at 19-3-14 下午6:47
* &＃64;version 1.0.0
*/
public class BatchReadFile {
private final Logger LOGGER &＃61; LoggerFactory.getLogger(BatchReadFile.class);
/**
* 字符集UTF-8
*/
public static final String CHARSET_UTF8 &＃61; "UTF-8";
/**
* 字符集GBK
*/
public static final String CHARSET_GBK &＃61; "GBK";
/**
* 字符集gb2312
*/
public static final String CHARSET_GB2312 &＃61; "gb2312";
/**
* 文件内容分割符&＃xff0d;逗号
*/
public static final String SEPARATOR_COMMA &＃61; ",";
private int bufSize &＃61; 1024;
// 换行符
private byte key &＃61; "\n".getBytes()[0];
// 当前行数
private long lineNum &＃61; 0;
// 文件编码,默认为gb2312
private String encode &＃61; CHARSET_GB2312;
// 具体业务逻辑监听器
private ReaderFileListener readerListener;
public void setEncode(String encode) {
this.encode &＃61; encode;
}
public void setReaderListener(ReaderFileListener readerListener) {
this.readerListener &＃61; readerListener;
}
/**
* 获取准确开始位置
* &＃64;param file
* &＃64;param position
* &＃64;return
* &＃64;throws Exception
*/
public long getStartNum(File file, long position) throws Exception {
long startNum &＃61; position;
FileChannel fcin &＃61; new RandomAccessFile(file, "r").getChannel();
fcin.position(position);
try {
int cache &＃61; 1024;
ByteBuffer rBuffer &＃61; ByteBuffer.allocate(cache);
// 每次读取的内容
byte[] bs &＃61; new byte[cache];
// 缓存
byte[] tempBs &＃61; new byte[0];
while (fcin.read(rBuffer) !&＃61; -1) {
int rSize &＃61; rBuffer.position();
rBuffer.rewind();
rBuffer.get(bs);
rBuffer.clear();
byte[] newStrByte &＃61; bs;
// 如果发现有上次未读完的缓存,则将它加到当前读取的内容前面
if (null !&＃61; tempBs) {
int tL &＃61; tempBs.length;
newStrByte &＃61; new byte[rSize &＃43; tL];
System.arraycopy(tempBs, 0, newStrByte, 0, tL);
System.arraycopy(bs, 0, newStrByte, tL, rSize);
}
// 获取开始位置之后的第一个换行符
int endIndex &＃61; indexOf(newStrByte, 0);
if (endIndex !&＃61; -1) {
return startNum &＃43; endIndex;
}
tempBs &＃61; substring(newStrByte, 0, newStrByte.length);
startNum &＃43;&＃61; 1024;
}
} finally {
fcin.close();
}
return position;
}
/**
* 从设置的开始位置读取文件&＃xff0c;一直到结束为止。如果 end设置为负数,刚读取到文件末尾
* &＃64;param fullPath
* &＃64;param start
* &＃64;param end
* &＃64;throws Exception
*/
public void readFileByLine(String fullPath, long start, long end) throws Exception {
File fin &＃61; new File(fullPath);
if (!fin.exists()) {
throw new FileNotFoundException("没有找到文件&＃xff1a;" &＃43; fullPath);
}
FileChannel fileChannel &＃61; new RandomAccessFile(fin, "r").getChannel();
fileChannel.position(start);
try {
ByteBuffer rBuffer &＃61; ByteBuffer.allocate(bufSize);
// 每次读取的内容
byte[] bs &＃61; new byte[bufSize];
// 缓存
byte[] tempBs &＃61; new byte[0];
String line;
// 当前读取文件位置
long nowCur &＃61; start;
while (fileChannel.read(rBuffer) !&＃61; -1) {
int rSize &＃61; rBuffer.position();
rBuffer.rewind();
rBuffer.get(bs);
rBuffer.clear();
byte[] newStrByte;
//去掉表头
if(nowCur &＃61;&＃61; start){
int firstLineIndex &＃61; indexOf(bs, 0);
int newByteLenth &＃61; bs.length-firstLineIndex-1;
newStrByte &＃61; new byte[newByteLenth];
System.arraycopy(bs, firstLineIndex&＃43;1, newStrByte, 0, newByteLenth);
} else {
newStrByte &＃61; bs;
}
// 如果发现有上次未读完的缓存,则将它加到当前读取的内容前面
if (null !&＃61; tempBs && tempBs.length !&＃61; 0) {
int tL &＃61; tempBs.length;
newStrByte &＃61; new byte[rSize &＃43; tL];
System.arraycopy(tempBs, 0, newStrByte, 0, tL);
System.arraycopy(bs, 0, newStrByte, tL, rSize);
}
// 是否已经读到最后一位
boolean isEnd &＃61; false;
nowCur &＃43;&＃61; bufSize;
// 如果当前读取的位数已经比设置的结束位置大的时候&＃xff0c;将读取的内容截取到设置的结束位置
if (end > 0 && nowCur > end) {
// 缓存长度 - 当前已经读取位数 - 最后位数
int l &＃61; newStrByte.length - (int) (nowCur - end);
newStrByte &＃61; substring(newStrByte, 0, l);
isEnd &＃61; true;
}
int fromIndex &＃61; 0;
int endIndex &＃61; 0;
// 每次读一行内容&＃xff0c;以 key(默认为\n) 作为结束符
while ((endIndex &＃61; indexOf(newStrByte, fromIndex)) !&＃61; -1) {
byte[] bLine &＃61; substring(newStrByte, fromIndex, endIndex);
line &＃61; new String(bLine, 0, bLine.length, encode);
lineNum&＃43;&＃43;;
// 输出一行内容&＃xff0c;处理方式由调用方提供
readerListener.outLine(line.trim(), lineNum, false);
fromIndex &＃61; endIndex &＃43; 1;
}
// 将未读取完成的内容放到缓存中
tempBs &＃61; substring(newStrByte, fromIndex, newStrByte.length);
if (isEnd) {
break;
}
}
// 将剩下的最后内容作为一行&＃xff0c;输出&＃xff0c;并指明这是最后一行
String lineStr &＃61; new String(tempBs, 0, tempBs.length, encode);
readerListener.outLine(lineStr.trim(), lineNum, true);
} finally {
fileChannel.close();
fin.deleteOnExit();
}
}
/**
* 查找一个byte[]从指定位置之后的一个换行符位置
*
* &＃64;param src
* &＃64;param fromIndex
* &＃64;return
* &＃64;throws Exception
*/
private int indexOf(byte[] src, int fromIndex) throws Exception {
for (int i &＃61; fromIndex; i
if (src[i] &＃61;&＃61; key) {
return i;
}
}
return -1;
}
/**
* 从指定开始位置读取一个byte[]直到指定结束位置为止生成一个全新的byte[]
*
* &＃64;param src
* &＃64;param fromIndex
* &＃64;param endIndex
* &＃64;return
* &＃64;throws Exception
*/
private byte[] substring(byte[] src, int fromIndex, int endIndex) throws Exception {
int size &＃61; endIndex - fromIndex;
byte[] ret &＃61; new byte[size];
System.arraycopy(src, fromIndex, ret, 0, size);
return ret;
}
}
以上是关键代码&＃xff1a;利用FileChannel与ByteBuffer从磁盘中分批读取数据
多线程调用批量读取&＃xff1a;
/**
* 类功能描述: 线程读取文件
*
* &＃64;author WangXueXing create at 19-3-14 下午6:51
* &＃64;version 1.0.0
*/
public class ReadFileThread extends Thread {
private ReaderFileListener processDataListeners;
private String filePath;
private long start;
private long end;
private Thread preThread;
public ReadFileThread(ReaderFileListener processDataListeners,
long start,long end,
String file) {
this(processDataListeners, start, end, file, null);
}
public ReadFileThread(ReaderFileListener processDataListeners,
long start,long end,
String file,
Thread preThread) {
this.setName(this.getName()&＃43;"-ReadFileThread");
this.start &＃61; start;
this.end &＃61; end;
this.filePath &＃61; file;
this.processDataListeners &＃61; processDataListeners;
this.preThread &＃61; preThread;
}
&＃64;Override
public void run() {
BatchReadFile readFile &＃61; new BatchReadFile();
readFile.setReaderListener(processDataListeners);
readFile.setEncode(processDataListeners.getEncode());
try {
readFile.readFileByLine(filePath, start, end &＃43; 1);
if(this.preThread !&＃61; null){
this.preThread.join();
}
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
监听读取&＃xff1a;
import java.util.ArrayList;
import java.util.List;
/**
* 类功能描述&＃xff1a;读文件监听父类
*
* &＃64;author WangXueXing create at 19-3-14 下午6:52
* &＃64;version 1.0.0
*/
public abstract class ReaderFileListener
// 一次读取行数&＃xff0c;默认为1000
private int readColNum &＃61; 1000;
/**
* 文件编码
*/
private String encode;
/**
* 分批读取行列表
*/
private List
/**
*其他参数
*/
private T otherParams;
/**
* 每读取到一行数据&＃xff0c;添加到缓存中
* &＃64;param lineStr 读取到的数据
* &＃64;param lineNum 行号
* &＃64;param over 是否读取完成
* &＃64;throws Exception
*/
public void outLine(String lineStr, long lineNum, boolean over) throws Exception {
if(null !&＃61; lineStr && !lineStr.trim().equals("")){
rowList.add(lineStr);
}
if (!over && (lineNum % readColNum &＃61;&＃61; 0)) {
output(rowList);
rowList &＃61; new ArrayList<>();
} else if (over) {
output(rowList);
rowList &＃61; new ArrayList<>();
}
}
/**
* 批量输出
*
* &＃64;param stringList
* &＃64;throws Exception
*/
public abstract void output(List
/**
* 设置一次读取行数
* &＃64;param readColNum
*/
protected void setReadColNum(int readColNum) {
this.readColNum &＃61; readColNum;
}
public String getEncode() {
return encode;
}
public void setEncode(String encode) {
this.encode &＃61; encode;
}
public T getOtherParams() {
return otherParams;
}
public void setOtherParams(T otherParams) {
this.otherParams &＃61; otherParams;
}
public List
return rowList;
}
public void setRowList(List
this.rowList &＃61; rowList;
}
}
实现监听读取并分批调用插入数据接口&＃xff1a;
import com.today.api.finance.ImportServiceClient;
import com.today.api.finance.request.ImportRequest;
import com.today.api.finance.response.ImportResponse;
import com.today.api.finance.service.ImportService;
import com.today.common.Constants;
import com.today.domain.StaffSimpInfo;
import com.today.util.EmailUtil;
import com.today.util.UserSessionHelper;
import com.today.util.readfile.ReadFile;
import com.today.util.readfile.ReadFileThread;
import com.today.util.readfile.ReaderFileListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import java.io.File;
import java.io.FileInputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.FutureTask;
import java.util.stream.Collectors;
/**
* 类功能描述&＃xff1a;报表导入服务实现
*
* &＃64;author WangXueXing create at 19-3-19 下午1:43
* &＃64;version 1.0.0
*/
&＃64;Service
public class ImportReportServiceImpl extends ReaderFileListener
private final Logger LOGGER &＃61; LoggerFactory.getLogger(ImportReportServiceImpl.class);
&＃64;Value("${READ_COL_NUM_ONCE}")
private String readColNum;
&＃64;Value("${REPORT_IMPORT_RECEIVER}")
private String reportImportReceiver;
/**
* 财务报表导入接口
*/
private ImportService service &＃61; new ImportServiceClient();
/**
* 读取文件内容
* &＃64;param file
*/
public void readTxt(File file, ImportRequest importRequest) throws Exception {
this.setOtherParams(importRequest);
ReadFile readFile &＃61; new ReadFile();
try(FileInputStream fis &＃61; new FileInputStream(file)){
int available &＃61; fis.available();
long maxThreadNum &＃61; &＃xff13;L;
// 线程粗略开始位置
long i &＃61; available / maxThreadNum;
this.setRowList(new ArrayList<>());
StaffSimpInfo staffSimpInfo &＃61; ((StaffSimpInfo)UserSessionHelper.getCurrentUserInfo().getData());
String finalReportReceiver &＃61; getEmail(staffSimpInfo.getEmail(), reportImportReceiver);
this.setReadColNum(Integer.parseInt(readColNum));
this.setEncode(ReadFile.CHARSET_GB2312);
//这里单独使用一个线程是为了当maxThreadNum大于1的时候&＃xff0c;统一管理这些线程
new Thread(()->{
Thread preThread &＃61; null;
FutureTask futureTask &＃61; null ;
try {
for (long j &＃61; 0; j
//计算精确开始位置
long startNum &＃61; j &＃61;&＃61; 0 ? 0 : readFile.getStartNum(file, i * j);
long endNum &＃61; j &＃43; 1
//具体监听实现
preThread &＃61; new ReadFileThread(this, startNum, endNum, file.getPath(), preThread);
futureTask &＃61; new FutureTask(preThread, new Object());
futureTask.run();
}
if(futureTask.get() !&＃61; null) {
EmailUtil.sendEmail(EmailUtil.REPORT_IMPORT_EMAIL_PREFIX, finalReportReceiver, "导入报表成功", "导入报表成功" ); //todo 等文案
}
} catch (Exception e){
futureTask.cancel(true);
try {
EmailUtil.sendEmail(EmailUtil.REPORT_IMPORT_EMAIL_PREFIX, finalReportReceiver, "导入报表失败", e.getMessage());
} catch (Exception e1){
//ignore
LOGGER.error("发送邮件失败", e1);
}
LOGGER.error("导入报表类型:"&＃43;importRequest.getReportType()&＃43;"失败", e);
} finally {
futureTask.cancel(true);
}
}).start();
}
}
private String getEmail(String infoEmail, String reportImportReceiver){
if(StringUtils.isEmpty(infoEmail)){
return reportImportReceiver;
}
return infoEmail;
}
/**
* 每批次调用导入接口
* &＃64;param stringList
* &＃64;throws Exception
*/
&＃64;Override
public void output(List
ImportRequest importRequest &＃61; this.getOtherParams();
List
.map(x->Arrays.asList(x.split(ReadFile.SEPARATOR_COMMA)).stream().map(String::trim).collect(Collectors.toList()))
.collect(Collectors.toList());
LOGGER.info("上传数据:{}", dataList);
importRequest.setDataList(dataList);
// LOGGER.info("request对象&＃xff1a;{}",importRequest, "request增加请求字段&＃xff1a;{}", importRequest.data);
ImportResponse importResponse &＃61; service.batchImport(importRequest);
LOGGER.info("&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;SUCESS_CODE&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;"&＃43;importResponse.getCode());
//导入错误,输出错误信息
if(!Constants.SUCESS_CODE.equals(importResponse.getCode())){
LOGGER.error("导入报表类型:"&＃43;importRequest.getReportType()&＃43;"失败","返回码为&＃xff1a;", importResponse.getCode() ,"返回信息&＃xff1a;",importResponse.getMessage());
throw new RuntimeException("导入报表类型:"&＃43;importRequest.getReportType()&＃43;"失败"&＃43;"返回码为&＃xff1a;"&＃43; importResponse.getCode() &＃43;"返回信息&＃xff1a;"&＃43;importResponse.getMessage());
}
// if(importResponse.data !&＃61; null && importResponse.data.get().get("batchImportFlag")!&＃61;null) {
// LOGGER.info("eywa-service请求batchImportFlag不为空");
// }
importRequest.setData(importResponse.data);
}
}
注意&＃xff1a;第53行代码&＃xff1a;long maxThreadNum &＃61; &＃xff13;L;
就是设置分批读取磁盘文件的线程数&＃xff0c;我设置为&＃xff13;&＃xff0c;大家不要设置太大&＃xff0c;不然多个线程读取到内存&＃xff0c;也会造成服务器内存溢出。
以上所有批次的批量读取并调用插入接口都成功发送邮件通知给导入人&＃xff0c;任何一个批次失败直接发送失败邮件。
数据库分批插入数据&＃xff1a;
/**
* 批量插入非联机第三方导入账单
* &＃64;param dataList
*/
def insertNonOnlinePayment(dataList: List[NonOnlineSourceData]) : Unit &＃61; {
if (dataList.nonEmpty) {
CheckAccountDataSource.mysqlData.withConnection { conn &＃61;>
val sql &＃61;
s""" INSERT INTO t_pay_source_data
(store_code,
store_name,
source_date,
order_type,
trade_type,
third_party_payment_no,
business_type,
business_amount,
trade_time,
created_at,
updated_at)
VALUES (?,?,?,?,?,?,?,?,?,NOW(),NOW())"""
conn.setAutoCommit(false)
var stmt &＃61; conn.prepareStatement(sql)
var i &＃61; 0
dataList.foreach { x &＃61;>
stmt.setString(1, x.storeCode)
stmt.setString(2, x.storeName)
stmt.setString(3, x.sourceDate)
stmt.setInt(4, x.orderType)
stmt.setInt(5, x.tradeType)
stmt.setString(6, x.tradeNo)
stmt.setInt(7, x.businessType)
stmt.setBigDecimal(8, x.businessAmount.underlying())
stmt.setString(9, x.tradeTime.getOrElse(null))
stmt.addBatch()
if ((i % 5000 &＃61;&＃61; 0) && (i !&＃61; 0)) { //分批提交
stmt.executeBatch
conn.commit
conn.setAutoCommit(false)
stmt &＃61; conn.prepareStatement(sql)
}
i &＃43;&＃61; 1
}
stmt.executeBatch()
conn.commit()
}
}
}
以上代码实现每5000 行提交一次批量插入&＃xff0c;防止一次提较数据库的压力。
☆
往期精彩
☆
01 漫谈发版哪些事&＃xff0c;好课程推荐
02 Linux的常用最危险的命令
03 精讲Spring Boot—入门&＃43;进阶&＃43;实例
04 优秀的Java程序员必须了解的GC哪些
05 互联网支付系统整体架构详解
关注我
每天进步一点点
京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有