多线程往Oracle数据库里插入数据的优化

xiangxingchina

浏览: 507790 次
性别:
来自: 北京

最近访客更多访客>>

lanmubai

forfelicity

linktoyl22

wuweilue

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

BUG解决备忘录

多线程 Oracle thread 网络应用 Apache

昨天做了一个东西，要实现解析txt文件，然后入库的功能。开始试验了一下单线程插入，速度实在太慢了，半个小时才插入了2W多条数据，后来改用Java的线程池启用了30个线程，并发的执行，插入100W条数据用了一个多小时。后来又对业务层的事务做了一些调整，每1000条insert之后才提交一次，一共开了20个线程，最后100W条数据入库一共用了14分钟不到，平均一分钟7.1W条左右。代码如下：

/**
* 分析Apache日志的定时任务.每天运行数次.
*
* @author <a href="mailto:HL_Qu@hotmail.com">Along</a>
*
* @version $Revision$
*
* @since 2009-2-9
*/
public class ApacheLogAnalysisTask {
/**
* Logger for this class
*/
private static final Log logger = LogFactory.getLog(ApacheLogAnalysisTask.class);
//总线程数
private static final int THREAD_COUNT = 20;
//每个线程插入的日志数
private static final long LOG_COUNT_PER_THREAD = 1000;
//日志文件的位置
private static final String LOG_FILE = Property.LOG_FILE_PATH + "formatLog.txt";
private IObjectActionDetailService objectActionDetailService;
public void setObjectActionDetailService(IObjectActionDetailService objectActionDetailService) {
this.objectActionDetailService = objectActionDetailService;
}
public void execute() {
this.multiAnalysisLog();
}
private void multiAnalysisLog() {
ExecutorService exePool = Executors.newFixedThreadPool(THREAD_COUNT);
FileReader fr = null;
BufferedReader br = null;
long beginLine = 1;
long endLine = 0;
String logFileBack = Property.LOG_FILE_PATH + "/formatLog_" + DateUtil.getSystemCurrentDate() + "_" + System.currentTimeMillis() + ".txt";
try {
//文件拷贝出来一个新的,并将原先的删除.
FileUtil.copyfile(LOG_FILE, logFileBack, true);
FileUtil.deleteFile(LOG_FILE);
System.out.println(logFileBack);
fr = new FileReader(logFileBack);
br = new BufferedReader(fr);
while ((br.readLine()) != null){
endLine++;
//每个线程分配固定的行数
if((endLine - beginLine + 1) == LOG_COUNT_PER_THREAD) {
exePool.execute(new AnalysisLogTask(logFileBack, beginLine, endLine));
beginLine = endLine + 1;
}
}
//最后一个线程
if (endLine > beginLine) {
exePool.execute(new AnalysisLogTask(logFileBack, beginLine, endLine));
}
} catch (FileNotFoundException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
} finally {
if (br != null) {
try {
br.close();
br = null;
} catch (IOException e) {
if (logger.isErrorEnabled()) {
logger.error("run()", e);
}
e.printStackTrace();
}
}
if (fr != null) {
try {
fr.close();
fr = null;
} catch (IOException e) {
if (logger.isErrorEnabled()) {
logger.error("run()", e);
}
e.printStackTrace();
}
}
exePool.shutdown();
while (true) {
if (exePool.isTerminated()) {
System.out.println("ShutDown");
FileUtil.deleteFile(logFileBack);
break;
}
}
}
}
private class AnalysisLogTask implements Runnable {
//起始行
private long beginLine;
//结束行
private long endLine;
private String logFilePath;
public AnalysisLogTask(String logFilePath, long beginLine, long endLine) {
super();
this.logFilePath = logFilePath;
this.beginLine = beginLine;
this.endLine = endLine;
}
@Override
public void run() {
FileReader fr = null;
BufferedReader br = null;
String tempStr = null;
String[] tempArray = null;
long currentLine = 0;
List <ObjectActionDetail> resultList = new ArrayList<ObjectActionDetail>();
ObjectActionDetail tempObjectActionDetailVO = null;
try {
fr = new FileReader(logFilePath);
br = new BufferedReader(fr);
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//跳过前置的行数
if (beginLine != 1) {
while ((currentLine < (beginLine-1)) && br.readLine() != null) {
++currentLine;
}
}
while ((tempStr = br.readLine()) != null && currentLine++ < endLine) {
tempArray = tempStr.split("\t");
tempObjectActionDetailVO = new ObjectActionDetail();
tempObjectActionDetailVO.setIp(tempArray[0]);
tempObjectActionDetailVO.setActionTime(sdf.parse(tempArray[1]));
tempObjectActionDetailVO.setSrcObjTypeId(Integer.parseInt(tempArray[2]));
tempObjectActionDetailVO.setSrcObjId(Integer.parseInt(tempArray[3]));
tempObjectActionDetailVO.setTarObjTypeId(Integer.parseInt(tempArray[4]));
tempObjectActionDetailVO.setTarObjId(Integer.parseInt(tempArray[5]));
tempObjectActionDetailVO.setActionId(Integer.parseInt(tempArray[6]));
tempObjectActionDetailVO.setScore(0);
tempObjectActionDetailVO.setStatus(1);
resultList.add(tempObjectActionDetailVO);
}
logger.info("Thread:" + Thread.currentThread().getName() + " beginLine=" + beginLine + " endLine=" + endLine);
objectActionDetailService.insertObjectActionDetailBatch(resultList);
} catch (FileNotFoundException e) {
if (logger.isErrorEnabled()) {
logger.error("run()", e);
}
e.printStackTrace();
} catch (IOException e) {
if (logger.isErrorEnabled()) {
logger.error("run()", e);
}
e.printStackTrace();
} catch (ParseException e) {
if (logger.isErrorEnabled()) {
logger.error("run()", e);
}
e.printStackTrace();
} finally {
if (br != null) {
try {
br.close();
br = null;
} catch (IOException e) {
if (logger.isErrorEnabled()) {
logger.error("run()", e);
}
e.printStackTrace();
}
}
if (fr != null) {
try {
fr.close();
fr = null;
} catch (IOException e) {
if (logger.isErrorEnabled()) {
logger.error("run()", e);
}
e.printStackTrace();
}
}
}
}
}
}

今天又试验了一下先将整个文件拆分成小的单个文件，然后每个线程再解析自己的文件。测试后感觉这样的效果不好，明显不如多线程从一个文章读数据好，27分钟插入了100W条数据,平均一分钟3.7W条左右。怀疑是多线程读取文件，本地磁盘的I/O受限导致性能低下。代码如下:

/**
* 分析Apache日志的定时任务.每天运行数次.
*
* @author <a href="mailto:HL_Qu@hotmail.com">Along</a>
*
* @version $Revision$
*
* @since 2009-2-9
*/
public class ApacheLogAnalysisTask {
/**
* Logger for this class
*/
private static final Log logger = LogFactory.getLog(ApacheLogAnalysisTask.class);
//总线程数
private static final int THREAD_COUNT = 10;
//每个线程插入的日志数
private static final long LOG_COUNT_PER_THREAD = 3000;
//日志文件的位置
private static final String LOG_FILE = Property.LOG_FILE_PATH + "/formatLog.txt";
private IObjectActionDetailService objectActionDetailService;
public void setObjectActionDetailService(IObjectActionDetailService objectActionDetailService) {
this.objectActionDetailService = objectActionDetailService;
}
public void execute() {
this.multiAnalysisLog();
}
private void multiAnalysisLog() {
ExecutorService exePool = Executors.newFixedThreadPool(THREAD_COUNT);
FileReader fr = null;
FileWriter fw = null;
BufferedReader br = null;
int threadCount = 0;
long tempLineCount = 0;
String tempReadLineStr = null;
long now = System.currentTimeMillis();
String logFileBackFile = Property.LOG_FILE_PATH + "/old/formatLog_" + DateUtil.getSystemCurrentDate() + "_" + now + ".txt";
String logFilePerThreadName = Property.LOG_FILE_PATH + "/old/formatLog_" + DateUtil.getSystemCurrentDate() + "_" + now;
try {
//文件拷贝出来一个新的,并将原先的删除.
FileUtil.copyfile(LOG_FILE, logFileBackFile, true);
//FileUtil.deleteFile(LOG_FILE);
fr = new FileReader(logFileBackFile);
br = new BufferedReader(fr);
fw = new FileWriter(logFilePerThreadName + "_" + ++threadCount + ".txt");
while ((tempReadLineStr = br.readLine()) != null){
tempLineCount++;
fw.append(tempReadLineStr).append("\r\n");
//每个线程分配固定的行数
if(tempLineCount == LOG_COUNT_PER_THREAD) {
fw.flush();
fw.close();
exePool.execute(new AnalysisLogTask(logFilePerThreadName + "_" + threadCount + ".txt"));
//创建新的文件,临时变量清零.
fw = new FileWriter(logFilePerThreadName + "_" + ++threadCount + ".txt");
tempLineCount = 0;
}
}
//最后一个线程有文件则写入执行,没有,则删除最后一个建立的文件.
if (tempLineCount != 0) {
fw.flush();
fw.close();
exePool.execute(new AnalysisLogTask(logFilePerThreadName + "_" + threadCount + ".txt"));
} else {
fw.flush();
fw.close();
FileUtil.deleteFile(logFilePerThreadName + "_" + threadCount + ".txt");
}
} catch (FileNotFoundException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
} finally {
if (br != null) {
try {
br.close();
br = null;
} catch (IOException e) {
if (logger.isErrorEnabled()) {
logger.error("run()", e);
}
e.printStackTrace();
}
}
if (fr != null) {
try {
fr.close();
fr = null;
} catch (IOException e) {
if (logger.isErrorEnabled()) {
logger.error("run()", e);
}
e.printStackTrace();
}
}
FileUtil.deleteFile(logFileBackFile);
logger.info("File has deleted:" + logFileBackFile);
exePool.shutdown();
//判断是不是所有的任务都执行完毕,执行完删除日志文件.
while (true) {
if (exePool.isTerminated()) {
logger.info("All task has shutdown.");
break;
}
}
}
}
private class AnalysisLogTask implements Runnable {
//每个线程要处理的日志文件
private String logFilePath;
public AnalysisLogTask(String logFilePath) {
super();
this.logFilePath = logFilePath;
}
@Override
public void run() {
logger.info("Thread:" + Thread.currentThread().getName() + " running.");
FileReader fr = null;
BufferedReader br = null;
String tempStr = null;
String[] tempArray = null;
List <ObjectActionDetail> resultList = new ArrayList<ObjectActionDetail>();
ObjectActionDetail tempObjectActionDetailVO = null;
try {
fr = new FileReader(logFilePath);
br = new BufferedReader(fr);
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
while ((tempStr = br.readLine()) != null) {
tempArray = tempStr.split("\t");
tempObjectActionDetailVO = new ObjectActionDetail();
tempObjectActionDetailVO.setIp(tempArray[0]);
tempObjectActionDetailVO.setActionTime(sdf.parse(tempArray[1]));
tempObjectActionDetailVO.setSrcObjTypeId(Integer.parseInt(tempArray[2]));
tempObjectActionDetailVO.setSrcObjId(Integer.parseInt(tempArray[3]));
tempObjectActionDetailVO.setTarObjTypeId(Integer.parseInt(tempArray[4]));
tempObjectActionDetailVO.setTarObjId(Integer.parseInt(tempArray[5]));
tempObjectActionDetailVO.setActionId(Integer.parseInt(tempArray[6]));
tempObjectActionDetailVO.setScore(0);
tempObjectActionDetailVO.setStatus(1);
resultList.add(tempObjectActionDetailVO);
}
objectActionDetailService.insertObjectActionDetailBatch(resultList);
} catch (FileNotFoundException e) {
if (logger.isErrorEnabled()) {
logger.error("run()", e);
}
e.printStackTrace();
} catch (IOException e) {
if (logger.isErrorEnabled()) {
logger.error("run()", e);
}
e.printStackTrace();
} catch (ParseException e) {
if (logger.isErrorEnabled()) {
logger.error("run()", e);
}
e.printStackTrace();
} finally {
if (br != null) {
try {
br.close();
br = null;
} catch (IOException e) {
if (logger.isErrorEnabled()) {
logger.error("run()", e);
}
e.printStackTrace();
}
}
if (fr != null) {
try {
fr.close();
fr = null;
} catch (IOException e) {
if (logger.isErrorEnabled()) {
logger.error("run()", e);
}
e.printStackTrace();
}
}
//删除本线程负责的日志文件
FileUtil.deleteFile(logFilePath);
logger.info("Thread:" + Thread.currentThread().getName() + " logFilePath has deleted:" + logFilePath);
logger.info("Thread:" + Thread.currentThread().getName() + " shutdown.");
}
}
}
}

后来又找系统管理员优化了一下网络，现在数据入库的速度是100W条7分钟 。相信应用和数据库移动到生产环境中，性能会进一步提升。

分享到：

Tomcat性能参数 | JSTL操作Map

2010-10-21 23:39
浏览 4530
评论(0)
分类:编程语言
查看更多

发表评论

您还没有登录,请您登录后再发表评论

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论

多线程往Oracle数据库里插入数据的优化

评论

发表评论

相关推荐

最近访客 更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论

多线程往Oracle数据库里插入数据的优化

评论

发表评论

相关推荐

One of the two will be used. Which one is undefined.

Eclipse不提示的解决办法

Hibernate-Validation的使用

web.xml配置错误页面不起作用

struts2 :Unable to load configuration. ……struts-default.xml:46:178异常解

jsp中EL表达式不能显示的解决方法

FreeMark常用内置函数使用说明

lucene 排序

设置tomcat内存

没有启动服务器服务,此时尚未创建共享资源的解决方法是什么

struts2+displaytag翻页时Inappropriate OGNL expression解决方案（solution）

Tomcat性能参数

JSTL操作Map

Quartz的cron表达式

IBatis的缓存

Hibernate中get和load方法的区别

java的堆与栈的区别

IBatis和Spring结合时的Blob乱码处理

【转】InnoDB锁定模式

经典SQL语句集锦

最近访客更多访客>>