大批量數據讀寫

需求

大約200W條數據，批量從mysql中讀取，然后根據主鍵再從hbase讀數據進行關聯，最后再update到數據庫中

同步解決方案

同步解決方案，也是最接近人腦思考順序的方案是，分頁mysql讀取id集合，每頁1k條數據，然后拿着idList批量從nosql的hbase中進行數據的獲取，進行數據的封裝，然后逐條更新到數據庫中。實驗結果表明，如果要完成這項工作，估計要10小時以上。

先做個簡單的優化，盡可能的降低io開銷，將逐條更新回數據庫修改成，延遲批量提交數據庫。這樣1k次io開銷縮減成1次。利用ibatis的批量提交特性，具體代碼如下

@Override
public void batchUpdate(final List<T> objectList) {
	this.getSqlMapClientTemplate().execute(new SqlMapClientCallback() {
		public Object doInSqlMapClient(SqlMapExecutor executor) throws SQLException {
			executor.startBatch();
			for (T tmp : objectList) {
				executor.update(sqlmapNamespace + ".update" + , tmp);
			}
			return executor.executeBatch();
		}
	});
}

這樣再次實驗后，發現跟新1k條數據從原來的10s以上降低到300ms左右，還是有着非常大的提升的。整體的1k的分頁任務完成，從原來的40s左右降低到1.5s左右。那么完成200w左右的數據，仍然需要接近1個小時，不能滿足業務期望

異步解決方案

IO的開銷，已經基本上沒辦法在低成本的角度去優化了。那么可以從cpu的角度進行提高，運行top命令后發現，cpu的java占比基本在%0.3以內。因此可以嘗試采用多線程的異步方案進行並發處理。起線程的代碼如下,起大約10個線程左右，起的太多，會造成數據庫連接數超出，導致數據庫連接異常

query.setCurrentPage(1);
query.setPageSize(100000);// 開10個線程左右，能覆蓋200W的數據
Integer totalInteger = rUserAlipayDAO.countByQuery(query);
query.setTotalItem(totalInteger);
do {
	try {
		GetDataThread thread = new GetDataThread(query.getStartRow(), query.getEndRow());
		Thread t = new Thread(thread);
		t.start();
	} catch (Throwable t) {
		logger.error("update  error", t);
	}
} while (query.nextPage());

GetDataThread 代碼如下,接受10W條左右的數據，進行任務的操作,構造函數如下,主要用於區分每個線程處理的起始位置

public GetDataThread (Integer startRow, Integer endRow){
	this.startRow=startRow;
	this.endRow=endRow;
}

下面是處理的run方法

@Override
public void run() {
	...
	query.setStartRow(startRow);
	query.setPageSize(1000);//每頁1K條數據
	logger.warn("do startRow"+startRow +" thread start...");
	Long startLong = System.currentTimeMillis();
	do{
		//TODO 進行相關的任務處理
		startRow= startRow+1000;
		query.setStartRow(startRow);
	}while(startRow<endRow);
	
	logger.warn("Thread startRow"+startRow +"  cost "+(System.currentTimeMillis()-startLong));
}

調整后，每個線程處理10W條數據，大概2分鍾左右完成，但是由於是10個線程同時開工，總體任務的執行時間基本控制在3min以內，完全滿足業務期望

總結

這個場景非常簡單，寫這篇文章的目的主要是找到一個案例，讓初學者了解如何去分析一段代碼存在的性能問題，並且如何針對這些問題進行代碼改進。