Java并发包中常用类小结(二)

package com.yhj.container.concurrent;
 
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.FutureTask;
import java.util.concurrent.locks.ReentrantLock;
 
/**
 * @Described：异步通知测试用例
 * @author YHJ create at 2012-4-14 上午11:31:26
 * @FileNmae com.yhj.container.concurrent.FutureTaskTestCase.java
 */
public class FutureTaskTestCase {
 
    //测试需求： 使用一个key-value 形式存储
    //测试要求： 所有的连接池对象只能创建一次 且不能加载时初始化 在第一次访问时初始化目标连接池对象
    //测试实现： 通过HashMap加锁实现和FutureTask实现
 
    //Map测试任务 用例
    interface MapTask{
       //根据指定的key 获取指定的DB连接  key etc：mysql sqlserver oracle DB2 and so on
       public Connection getConnection(String key);
    }
 
    //枚举 数据库类型
    enum DB_TYPE{
       MYSQL(),SQLSERVR,ORACLE,DB2;
    }
 
    //使用HashMap加锁实现
    class HashMapWithLock implements MapTask{
 
       private Map<String, DBConnectionPool> pools = new HashMap<String, DBConnectionPool>();
 
       private ReentrantLock lock = new ReentrantLock();
 
       //加锁获取连接对象，防止高并发下数据重复创建
       @Override
       public Connection getConnection(String key){
           try {
              lock.lock(); //锁定操作，后续再来等待
              if(!pools.containsKey(key))
                  pools.put(key, new DBConnectionPool(key));
              return pools.get(key).getConnection();
           } finally{
              lock.unlock();//解锁操作
           }
       }
    }
 
    //使用ConcurrentHashMap实现，因为ConcurrentHashMap读取时不加锁，因此需要通过回调的方式控制并发
    class ConcurrentHashMapWithFutureTask implements MapTask{
 
       private ConcurrentHashMap<String, FutureTask<DBConnectionPool>> pools = new ConcurrentHashMap<String, FutureTask<DBConnectionPool>>();
 
       private FutureTask<DBConnectionPool> futureTask;
 
       //通过回调的方式 确保多线程下不会引发多次创建
       @Override
       public Connection getConnection(final String key){
           try {
              if(!pools.containsKey(key)){
                  Callable<DBConnectionPool> callable = new Callable<DBConnectionPool>() {
 
                     @Override
                     public DBConnectionPool call() throws Exception {
                         pools.put(key,futureTask);
                         return new DBConnectionPool(key);
                     }
                  };
                  FutureTask<DBConnectionPool> tmpTask = new FutureTask<DBConnectionPool>(callable);
                  futureTask = pools.putIfAbsent(key, tmpTask);
                  if(futureTask==null){
                     futureTask = tmpTask;
                     futureTask.run();
                  }
              }
              return pools.get(key).get().getConnection();
           } catch (Exception e) {
              e.printStackTrace();
              return null;
           }
       }
    }
 
    //DB连接池  测试用例供体
    class DBConnectionPool{
 
       public DBConnectionPool(String key) {
           System.out.println("创建了"+key+"类型的数据库连接池");
       }
 
       //获取DB连接
       public Connection getConnection(){
           // create Connection for db
           return new Connection();
       }
    }
 
    //DB连接 测试供体
    class Connection{
    }
 
    //任务执行器 待执行的任务
    class ExecutorTask implements Runnable{
 
       private CyclicBarrier barrier;//计数器
 
       private CountDownLatch latch;//集合点
 
       private MapTask task;//待执行的任务
 
       private String key;
 
       public ExecutorTask(String key,CyclicBarrier barrier, CountDownLatch latch,MapTask task) {
           this.barrier = barrier;
           this.latch = latch;
           this.task = task;
           this.key=key;
       }
 
       @Override
       public void run() {
           try {
              barrier.await();//到达集合点之前等待 确保数据是并发执行的
              Connection connection = task.getConnection(key);
              if(null==connection)
                  throw new NullPointerException("Null Connection Exception with "+key);
              latch.countDown();
           } catch (Exception e) {
              e.printStackTrace();
           }
       }
 
    }
 
    //执行函数
    public void execute(String key,int thredCount,MapTask task){
       CyclicBarrier barrier = new CyclicBarrier(thredCount);
       CountDownLatch latch = new CountDownLatch(thredCount);
       long beginTime = System.currentTimeMillis();
       System.out.println("===start "+task.getClass()+"===");
       for(int i=0;i<thredCount;++i){
           new Thread(new ExecutorTask(key, barrier, latch, task)).start();
       }
       try {
           latch.await();
           System.out.println("====end "+task.getClass()+" still time "+(System.currentTimeMillis()-beginTime)+"===");
       } catch (InterruptedException e1) {
           throw new RuntimeException(e1);
       }
    }
 
    //启动函数
    public void start(){
       int thredCount = 200;
       MapTask hashMapWithLock = new HashMapWithLock();
       MapTask concurrentHashMapWithFutureTask = new ConcurrentHashMapWithFutureTask();
       execute("mysql",thredCount, hashMapWithLock);
       execute("sqlserver",thredCount, concurrentHashMapWithFutureTask);
    }
 
    //主函数
    public static void main(String[] args) {
       //启动主进程
       new FutureTaskTestCase().start();
       //等待所有进程结束
       while(Thread.activeCount()>1){
           Thread.yield();
       }
    }
 
}