posts - 101,  comments - 29,  trackbacks - 0
并行计算框架的Java实现--系列三

优化锁，之前的锁是采用一个static的Object实现的，这要就会有一个问题，如果我创建了多个Executer，那么所有Job都会持有一把锁，既影响性能，也容易出现死锁的情况。所以，改成每个Executer持有一把锁。

Executer代码如下：

 

Java代码  收藏代码
  1. public class Executer {  
  2.     //计算已经派发的任务数(条件谓词)  
  3.     public static int THREAD_COUNT = 0;  
  4.     //存储任务的执行结果  
  5.     private List<Future<Object>> futres = new ArrayList<Future<Object>>();   
  6.     //条件队列锁  
  7.     public final Object lock = new Object();  
  8.     //线程池  
  9.     private ExecutorService pool = null;  
  10.     public Executer() {  
  11.         this(1);  
  12.     }  
  13.     public Executer(int threadPoolSize) {  
  14.         pool = Executors.newFixedThreadPool(threadPoolSize);  
  15.     }  
  16.     /** 
  17.      * 任务派发 
  18.      * @param job 
  19.      */  
  20.     public void fork(Job job){  
  21.         //设置同步锁  
  22.         job.setLock(lock);  
  23.         //将任务派发给线程池去执行  
  24.         futres.add(pool.submit(job));  
  25.         //增加线程数  
  26.         synchronized (lock) {  
  27.             THREAD_COUNT++;  
  28.         }  
  29.     }  
  30.     /** 
  31.      * 统计任务结果 
  32.      */  
  33.     public List<Object> join(){  
  34.         synchronized (lock) {  
  35.             while(THREAD_COUNT > 0){//检查线程数，如果为0，则表示所有任务处理完成  
  36.                 System.out.println("threadCount: "+THREAD_COUNT);  
  37.                 try {  
  38.                     lock.wait();//如果任务没有全部完成，则挂起。等待完成的任务给予通知  
  39.                 } catch (InterruptedException e) {  
  40.                     e.printStackTrace();  
  41.                 }  
  42.             }  
  43.         }  
  44.         List<Object> list = new ArrayList<Object>();  
  45.         //取出每个任务的处理结果，汇总后返回  
  46.         for (Future<Object> future : futres) {  
  47.             try {  
  48.                 Object result = future.get();//因为任务都已经完成，这里直接get  
  49.                 list.add(result);  
  50.             } catch (Exception e) {  
  51.                 e.printStackTrace();  
  52.             }   
  53.         }  
  54.         return list;  
  55.     }  
  56. }  

 Job类:

 

Java代码  收藏代码
  1. public abstract class Job implements Callable<Object> {  
  2.   
  3.     //锁  
  4.     private Object lock = null;  
  5.   
  6.     void setLock(Object lock) {  
  7.         this.lock = lock;  
  8.     }  
  9.   
  10.     @Override  
  11.     public Object call() throws Exception {  
  12.         Object result = this.execute();//执行子类具体任务  
  13.         synchronized (lock) {  
  14.             //处理完业务后，任务结束，递减线程数，同时唤醒主线程  
  15.             Executer.THREAD_COUNT--;  
  16.             lock.notifyAll();  
  17.         }  
  18.         return result;  
  19.     }  
  20.     /** 
  21.      * 业务处理函数 
  22.      */  
  23.     public abstract Object execute();  
  24.       
  25. }  

 测试结果：

 

Java代码  收藏代码
  1. threadCount: 10  
  2. running thread id = 8  
  3. running thread id = 10  
  4. running thread id = 9  
  5. running thread id = 12  
  6. running thread id = 11  
  7. threadCount: 8  
  8. threadCount: 7  
  9. threadCount: 6  
  10. threadCount: 5  
  11. running thread id = 12  
  12. running thread id = 8  
  13. running thread id = 11  
  14. threadCount: 2  
  15. running thread id = 10  
  16. threadCount: 1  
  17. running thread id = 9  
  18. ResultSize: 10  
  19. time: 2001  

 OK！

这样每个Executer就可以使用自己的lock，而相互不受同步的影响

posted on 2012-07-15 01:21 mixer-a 阅读(1097) 评论(0)  编辑  收藏

只有注册用户登录后才能发表评论。
网站导航:
 
Powered by:
BlogJava
Copyright ©2024 mixer-a