多线程实现数据库的并发操作

　　在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);
	}
}