多線程實現數據庫的並發操作

　　在Java中，程序需要操作數據庫，操作數據首要事就是要獲得數據庫的Connection對象，利用多線程對數據導入數據庫中將會加快操作進度，但是多個線程共享Connection對象，是不安全的，因為可以利用Java中的ThreadLocal為每個線程保存一個Connection對象，代碼如下：

package com.quar.innovation.db;

import java.sql.Connection;
import java.sql.DriverManager;

public class ConnnectionManager {

	private static final ThreadLocal<Connection> connectionHolder = new ThreadLocal<Connection>();
	
	private static final String BETADBURL = "jdbc:mysql://127.0.0.1:3306/test?useUnicode=true&characterEncoding=utf8&autoReconnect=true&user=root&password=root";

	
	public static Connection getConnectionFromThreadLocal() {
		Connection conn = connectionHolder.get();
		try {
			if (conn == null || conn.isClosed()) {
				Connection con = ConnnectionManager.getConnection();
				connectionHolder.set(con);
				System.out.println("[Thread]" + Thread.currentThread().getName());
				return con;
			}
			return conn;
		} catch (Exception e) {
			System.out.println("[ThreadLocal Get Connection Error]" + e.getMessage());
		}
		return null;
		
		
	}
	
	public static Connection getConnection() {
		Connection conn = null;
		try {
			Class.forName("com.mysql.jdbc.Driver");
			conn = (Connection) DriverManager.getConnection(BETADBURL);
		} catch (Exception e) {
			System.out.println("[Get Connection Error]" + e.getMessage());
		}
		return conn;
	}
}

　　通過ThreadLocal就可以為每個線程保留一份Connection對象，利用Java的ThreadPoolExecutor啟動線程池，完成數據庫操作，完整代碼如下：

public class QunarThreadPoolExecutor extends ThreadPoolExecutor {

    // 記錄每個線程執行任務開始時間
    private ThreadLocal<Long> start = new ThreadLocal<Long>();
    
    // 記錄所有任務完成使用的時間
    private AtomicLong totals = new AtomicLong();
    
    // 記錄線程池完成的任務數
    private AtomicInteger tasks = new AtomicInteger();
	
	public QunarThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
			BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
	}

	public QunarThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
			BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
	}

	public QunarThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
			BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
	}

	public QunarThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
			BlockingQueue<Runnable> workQueue) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
	}
	
	 /**
     * 每個線程在調用run方法之前調用該方法
     * */ 
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        start.set(System.currentTimeMillis());
    }

    /**
     * 每個線程在執行完run方法后調用該方法
     * */
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        tasks.incrementAndGet();
        totals.addAndGet(System.currentTimeMillis() - start.get());
    }

    @Override
    protected void terminated() {
        super.terminated();
        System.out.println("完成"+ tasks.get() +"個任務,平均耗時: [" + totals.get() / tasks.get() + "] ms");
    }


public class DataUpdater implements Runnable {

	private PreparedStatement pst;
	
	private List<UserProfileItem> userProfiles;
	
	private final String SQL = "insert into userprofile (`uid` ,`profile` , `logday`) VALUES (?, ? ,?) ON DUPLICATE KEY UPDATE `profile`= ? ";
	
	public DataUpdater(List<UserProfileItem> userProfiles) {
		this.userProfiles = userProfiles;
	}
	
	public void run() {
		try {
			pst = ConnnectionManager.getConnectionFromThreadLocal().prepareStatement(SQL);
			for (UserProfileItem userProfile : userProfiles) {
				if(userProfile.getUid() != null && !userProfile.getUid().isEmpty() && 
						userProfile.getProfile() != null && !userProfile.getProfile().isEmpty()) {
					pst.setString(1, userProfile.getUid());
					pst.setString(2, userProfile.getProfile());
					pst.setInt(3, userProfile.getLogday());
					pst.setString(4, userProfile.getProfile());
					pst.addBatch();
				}
			}
			pst.executeBatch();
		} catch (Exception e) {
			System.err.println("[SQL ERROR MESSAGE]" + e.getMessage());
		} finally {
			 close(pst);
		}
		
	}

	public void close(PreparedStatement pst) {
		if (pst != null) {
			try {
				pst.close();
			} catch (SQLException e) {
				System.err.println("[Close Statement Error]" + e.getMessage());
			}
		}
	}
}


public class UserProfileItem {

	private String uid;
	
	private String profile;
	
	private int logday;
	
	public UserProfileItem(String uid, String profile , int logday) {
		this.logday = logday;
		this.profile = profile;
		this.uid = uid;
	}

	public String getUid() {
		return uid;
	}

	public String getProfile() {
		return profile;
	}

	public int getLogday() {
		return logday;
	}
	
}

public class DataUpdaterMain {
	
	private LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
	
	private QunarThreadPoolExecutor qunarThreadPoolExecutor = new QunarThreadPoolExecutor(5, 8, 5, TimeUnit.MINUTES, queue);
	
	
	public void shutThreadPool(ThreadPoolExecutor executor) {
		if (executor != null) {
			executor.shutdown();
			try {
				if (!executor.awaitTermination(20 , TimeUnit.MINUTES)) {
					executor.shutdownNow();
				} 
			} catch (Exception e) {
				System.err.println("[ThreadPool Close Error]" + e.getMessage());
			}
			
		}
	}
	
	public void close(Reader reader) {
		if (reader != null) {
			try {
				reader.close();
			} catch (IOException e) {
				System.err.println("[Close Io Error]" + e.getMessage());
			}
		}
	}
	
	public void closeConnection(Connection conn , Statement st) {
		try {
			if (conn != null) {
				conn.close();
			}
			if (st != null) {
				conn.close();
			}
		} catch (Exception e) {
			System.err.println("[Close MySQL Error]" + e.getMessage());
		}
	}
	
	
	public boolean update(String file ,int logday) {
		long start = System.currentTimeMillis();
		BufferedReader br = null;
		int num = 0;
		try {
			br = new BufferedReader(new InputStreamReader(new FileInputStream(file)));
			String line = null;
			List<UserProfileItem> userProfiles = new LinkedList<UserProfileItem>();
			while ((line = br.readLine()) != null) {
				++num;
				String []items = line.split("\t");
				if (items.length == 2) {
					String uid = items[0];
					String profile = items[1];
					userProfiles.add(new UserProfileItem(uid, profile, logday));
					if (userProfiles.size() >= 100) {
						qunarThreadPoolExecutor.execute(new DataUpdater(userProfiles));
						userProfiles = new LinkedList<UserProfileItem>();
					}
				} else {
					System.err.println("[Data Error]" + line);
				}
			}
			qunarThreadPoolExecutor.execute(new DataUpdater(userProfiles));;
		} catch (Exception e) {
			e.printStackTrace();
			System.err.println("[Read File Error]" + e.getMessage());
			return false;
		}  finally {
			System.err.println("[Update] take time " + (System.currentTimeMillis() - start) + ".ms");
			System.err.println("[Update] update item " + num);
			shutThreadPool(qunarThreadPoolExecutor);;
			close(br);
		}
		return true;
	}
	
	public static void main(String []args) {
		String file = "D:\\workspaces\\promotionwordData.log";
		int logday = Integer.parseInt("20150606");
		DataUpdaterMain dataUpdaterMain = new DataUpdaterMain();
		dataUpdaterMain.update(file, logday);
	}
}