(文章本人原创，若转载请注明出处)
下面看一下一些具体的实现，先看一下Sender接口的commitData方法的MySql实现，即SenderMySqlImp类：
public void commitData(String path, String file) {
......
Connection connect = null;
try {
//根据配置由Helper来确定是否用连接池,这只调用getConnection()即可
connect = Helper.getConnection();
connect.setAutoCommit(false);
FileInputStream fis = new FileInputStream(path + file);
//insert语句,有三个参数分别为id,image,filename字段。
ps = connect.prepareStatement(sql.toString());
ps.setString(1, UUID.randomUUID().toString());
//将图片文件流化,用jdbc直接写到数据库
ps.setBinaryStream(2, fis, fis.available());
ps.setString(3, file);
ps.executeUpdate();
connect.commit();
count++;
Logger.writeLog("已处理文件数："+count+"，时间："+new java.util.Date());
} catch (Exception e) {
........
}
很简单吧，其实就是用Stream来做，另外在网上可以找到有关Oracle上传blob的实现，一般都是先insert一条记录，blob字段用empty_blob()函数插入一个空数据，然后再取出这个blob字段，最后按字节写入blob，具体参考SenderOracleImp类吧。个人感觉还是在mysql的这个效率高些并且看起来简单了很多。
然后来看看使用线程池的ProcessMulti类：
public class ProcessMulti implements Process{
private String path;
private Vector<String> files = new Vector<String>();
private Properties prop;
ProcessMulti() {
prop = ConfigMgr.getProperties(); //取config.properties中配置信息
this.path = prop.getProperty("path");
this.files = Helper.getFiles(prop.getProperty("filetype"), this.path);
}
public void doProcess() {
//正如前面两篇所说,这里是线程池构建器,传入相关参数
BlobSenderThreadPool tpe = new BlobSenderThreadPool(Integer
.valueOf(prop.getProperty("corePoolSize")), Integer
.valueOf(prop.getProperty("maxPoolSize")), Integer.valueOf(prop
.getProperty("keepAliveTime")), TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(Integer.valueOf(prop
.getProperty("maxPoolSize"))),
new BlobSenderThreadPool.DiscardPolicy(), files.size());
Logger.writeLogForce("开始处理
." + new java.util.Date());
for (int i = 0; i < this.files.size(); i++) {
//向线程池提交要处理的任务,线程池根据其配置进行处理
//Helper.getSender()会根据配置取得支持mysql或是oracel的写入方法
tpe.execute(new Runner(path, files.get(i), Helper.getSender()));
Logger.writeLog("已提交第" + (int)(i+1) + "个文件" + "，时间为："
+ new java.util.Date());
}
//可以在这里写一个打印输出,实际上程序很快就执行完上面的for,运行到这里,但是处理并没有完成,
//主程序好像职业经理人,他的工作就是分配任务给下属,自已就完成工作了,但下属们还要接着忙活呵呵...
System.out.println("任务已分配...");
}
//线程池类
class BlobSenderThreadPool extends ThreadPoolExecutor {
volatile int planTask;//计划任务,即计划要写的文件数
public BlobSenderThreadPool(int corePoolSize, int maximumPoolSize,
long keepAliveTime, TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler, int planTask) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
this.planTask = planTask;
}
@Override
//当某个线程处理完时会调用此方法
protected void afterExecute(Runnable r, Throwable t) {
Logger.writeLog("当前已完成任务数：" + (int)(this.getCompletedTaskCount()+1)
+ "计划任务数：" + planTask);
//若已处理完任务和计划的任务数相同说明所有线程已完成处理,终止线程池处理
if ((this.getCompletedTaskCount()+1)== planTask)
this.terminated();
super.afterExecute(r, t);
}
@Override
protected void terminated() {
Logger.writeLogForce("所有任务已完成 ,处理结束时间===>>" + new java.util.Date());
System.exit(0);
}
}
//要使用线程进行处理,类要实现Runable接口
class Runner implements Runnable {
String file;
String path;
Sender sender;
Runner(String path, String file, Sender sender) {
this.file = file;
this.path = path;
this.sender = sender;
}
//Runer的实例会被传入线程池,线程被运行时会调用run方法
public void run() {
sender.commitData(path, file);
}
}posted @
2009-03-21 10:40 依然Fantasy 阅读(616) |
评论 (0) |
编辑 收藏(文章本人原创，若转载请注明出处)
在实际当中的情况是系统数据库中需要上传大量照片到数据库中，数据量比较大，且不能在界面中通过操作逐个上传，要批量自动进行。其实起来也比较简单直接利用线程池将照片数据读取成流再存入BLOB字段即可。但是在实现后些功能后又进入了一些改造，实现了线程池、单线程、是否使用用连接池、不同数据库等不同的配置，这样在不同配置下可以观察到程序性能的不同。并且经过设计切换这些配置不需要修改程序。
使用DbAccess接口的getConnect()取得数据库连接，DbImp和DbPoolingImp实现了不使用连接池和使用连接池的两个版本。Sender接口的commitData()用来把BLOB数据写到数据库中，因为不同数据库可能写法有点不同所以这里SenderMySqlImp和SenderOracleImp分别是Mysql和Oracle的实现。Process接口的doProcess()是开始进行处理的方法，无论是单线程还是多线程。因此ProcessMulti和ProcessSingle是分别使用线程池以及单线程处理的类。ConfigMgr用于取得config.properties文件内配置信息，Logger是日志类，Helper中汇集了一些共用的静态方法。最后DataSender是主程序的类：）
posted @
2009-03-21 10:25 依然Fantasy 阅读(297) |
评论 (0) |
编辑 收藏(文章本人原创，若转载请注明出处)
在JDK1.5提供了一个线程池ThreadPoolExecutor，可以处理用户提交过来的线程。如果把要处理的任务比作盖一个大楼，那么每一个建筑工人就相当于一个线程，那么这个ThreadPoolExecutor就好像包工头，它来控制盖这个大楼需要多少个工人，何时招进新工人，何时辞退已经长时间没有事做的工人，等等此类事务。也就是说用户程序不断提交新的线程，ThreadPoolExecutor执行提交线程的同时会控制目前总共同时执行的线程数，销毁已执行完闲置的线程等控制行为，保留最少闲置线程数，并且可以配置不同的处理策略。
为什么要使用线程池呢，这与数据库连接池的原理有点相仿，线程的创建是需要成本的，包括服务器CPU和内存资源，由于多线程是并行运行，程序运行过程中可能有的线程已经完成自身处理任务，处于闲置状态，如果在这种情况下再不断创建新任务就是在浪费服务器资源，此时应该尽量使用先前创建的好的并且是处理闲置状态的线程来处理新任务，而线程池就可以有效的对此进行自动化管理，当然这个管理是可以由用户配置的。
ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler)
这是线程池的构建器，用户程序通过这个构建器传参数，corePoolSize是线程池中核心线程数，运行的线程数不能少于这个核心线程数，否则就新建线程。maximumPoolSize是充许最大的线程数。keepAliveTime设置除核心线程外其它线程的空闲时间，超过这个时间线程就自动终止。unit是指的keepAliveTime的时间单位。BlockingQueue接口按生产则消费者算法设计的一个线程池内部处理线程队列的接口，有三种实现SynchronousQueue、LinkedBlockingQueue和ArrayBlockingQueue，在实际运行程序时可以根据这三种实现得到不同的性能，比如有的实现可能在有新任务来时不新建线程，而是将其加入等待队列，等有线程运行完时再分配给其使用。具体实现还是参看它们的JDK文档吧，这里站在使用的角度它们是可以调整运行性能的开关。当最大线程和工作队列容量都达到最大值时，再提交给线程池新任务就会被拒绝，此时线程池会调用RejectedExecutionHandler 接口进行处理，具体实现有四种策略。我们只需要选用其中一种在构建ThreadPoolExecutor时传入即可。具体四种实现还是参看JDK文档吧。关于ThreadPoolExecutor的JDK文档。至此控制线程池运作的几个参数都从构建器中传入了。
posted @
2009-03-19 22:26 依然Fantasy 阅读(842) |
评论 (0) |
编辑 收藏