java實現分布式系統mysql的批量更新

背景

專案中開發乙個批量新增，更新的需求，由於之前的資料是一條一條入庫的，每天入庫量在十萬左右，後來增加需求每天的入庫量上公升到百萬時，入庫遭遇瓶頸，mq積壓嚴重。後來發現這樣需要頻繁的與mysql互動，且需要等待寫庫結果返回，效率堪憂拖慢其他模組，就有了批量新增和批量更新的需求。

優化

第一步：用執行緒池來更新，將更新**提交到執行緒池中，由執行緒池排程入庫

缺點：沒有解決與資料庫頻繁互動的問題。

第二步：執行模組不管更新結果，只需將更新任務放入乙個佇列中然後直接返回；用spring的定時任務註解@scheduled，指定乙個方法，隔一段時間呼叫一次入庫方法；入庫的邏輯是，獲取佇列當前任務個數cnt，迴圈poll任務然後新增到乙個list中，poll夠cnt個之後，通過批量更新方法將list更新到資料庫。

缺點：定時執行無法控制佇列大小，可能一次會取出很多條任務，也可能會把佇列撐得過大。

第三步：使用阻塞佇列放更新任務，用守護執行緒poll的佇列中的任務，當條數等於300條(此值根據實際情況，我們大佬建議300~500)，則批量更新一次，在poll時設定超時時間為2秒，當超過2秒還是沒有取到任務，則也批量把已經渠道的任務更新一次。

@service
public class batchexecutordatajob catch (interruptedexception e) 
}//初始化即呼叫
@postconstruct
private void init() ", new date(system.currenttimemillis()));
while (boolean.true) catch (interruptedexception e) 
if (null != poll) else 
// 如果任務list等於5000或poll超時且list中還有任務就批量更新
if (list.size() == 300|| 
(polltimeout && !collectionutils.isempty(list)))條任務,耗時{}毫秒", list.size(), 
system.currenttimemillis()-starttime);
list.clear();}}
});thread.setname("job-batchupdate-deamon");
// 設定啟動的執行緒為守護執行緒
thread.setdaemon(true);
thread.start();
}}

生產者方法就不說了，主要說消費者方法：

1、@postconstruct作用是在bean初始化之前執行消費者方法

2、poll = dataqueue.poll(2, timeunit.seconds)，使用阻塞佇列的poll方法，設定poll超時時間，當超過時間返回乙個null值

3、list.size() == 300，如果list中值等於300了就執行批量更新

4、polltimeout && !collectionutils.isempty(list)，如果poll超時了，說明當前生產者暫時沒有生產任務或不再生產任務，把list 中剩餘的任務批量更新

5、thread.setdaemon(true)，設定執行緒為守護執行緒，直到jvm停了才停止

最後提交**，檢視log日誌，使用批量入庫耗時比較長。查詢各種原因後jdbcurl 還需要新增配置

rewritebatchedstatements=true

新增上後重新部署專案，入庫耗時瞬間在1000ms以下了。

java實現分布式系統mysql的批量更新

JAVA實現分布式快取

JAVA實現分布式快取

分布式系統與Java技術

java實現分布式系統mysql的批量更新

JAVA實現分布式快取

JAVA實現分布式快取

分布式系統與Java技術

相關推薦