Java?多線程并發(fā)編程提高數(shù)據(jù)處理效率的詳細(xì)過程
??工作場景中遇到這樣一個(gè)需求：根據(jù)主機(jī)的 IP 地址聯(lián)動(dòng)更新其他模型的相關(guān)信息。需求很簡單，只涉及一般的數(shù)據(jù)庫聯(lián)動(dòng)查詢以及更新操作，然而在編碼實(shí)現(xiàn)過程中發(fā)現(xiàn)，由于主機(jī)的數(shù)量很多，導(dǎo)致循環(huán)遍歷查詢、更新時(shí)花費(fèi)很長的時(shí)間，調(diào)用一次接口大概需要 30-40 min 時(shí)間才能完成操作。
??因此，為了有效縮短接口方法的執(zhí)行時(shí)間，便考慮使用多線程并發(fā)編程方法，利用多核處理器并行執(zhí)行的能力，通過異步處理數(shù)據(jù)的方式，便可以大大縮短執(zhí)行時(shí)間，提高執(zhí)行效率。
??這里使用可重用固定線程數(shù)的線程池 FixedThreadPool，并使用 CountDownLatch 并發(fā)工具類提供的并發(fā)流程控制工具作為配合使用，保證多線程并發(fā)編程過程中的正常運(yùn)行：
- 首先，通過
Runtime.getRuntime().availableProcessors()方法獲取運(yùn)行機(jī)器的 CPU 線程數(shù)，用于后續(xù)設(shè)置固定線程池的線程數(shù)量。 - 其次，判斷任務(wù)的特性，如果為計(jì)算密集型任務(wù)則設(shè)置線程數(shù)為
CPU 線程數(shù)+1，如果為 IO 密集型任務(wù)則設(shè)置線程數(shù)為2 * CPU 線程數(shù)，由于在方法中需要與數(shù)據(jù)庫進(jìn)行頻繁的交互，因此屬于 IO 密集型任務(wù)。 - 之后，對(duì)數(shù)據(jù)進(jìn)行分組切割，每個(gè)線程處理一個(gè)分組的數(shù)據(jù)，分組的組數(shù)與線程數(shù)保持一致，并且還要?jiǎng)?chuàng)建計(jì)數(shù)器對(duì)象
CountDownLatch，調(diào)用構(gòu)造函數(shù)，初始化參數(shù)值為線程數(shù)個(gè)數(shù)，保證主線程等待所有子線程運(yùn)行結(jié)束后，再進(jìn)行后續(xù)的操作。 - 然后，調(diào)用
executorService.execute()方法，重寫run方法編寫業(yè)務(wù)邏輯與數(shù)據(jù)處理代碼，執(zhí)行完當(dāng)前線程后記得將計(jì)數(shù)器減1操作。 - 最后，當(dāng)所有子線程執(zhí)行完成后，關(guān)閉線程池。
?在省略工作場景中的業(yè)務(wù)邏輯代碼后，通用的處理方法示例如下所示：
public ResponseData updateHostDept() {
// ...
List<Map> hostMapList = mongoTemplate.find(query, Map.class, "host");
// split the hostMapList for the following multi-threads task
// return the number of logical CPUs
int processorsNum = Runtime.getRuntime().availableProcessors();
// set the threadNum as 2*(the number of logical CPUs) for handling IO Tasks,
// if Computing Tasks set the threadNum as (the number of logical CPUs) + 1
int threadNum = processorsNum * 2;
// the number of each group data
int eachGroupNum = hostMapList.size() / threadNum;
List<List<Map>> groupList = new ArrayList<>();
for (int i = 0; i < threadNum; i++) {
int start = i * eachGroupNum;
if (i == threadNum - 1) {
int end = mapList.size();
groupList.add(hostMapList.subList(start, end));
} else {
int end = (i+1) * eachGroupNum;
groupList.add(hostMapList.subList(start, end));
}
}
// update data by using multi-threads asynchronously
ExecutorService executorService = Executors.newFixedThreadPool(threadNum/2);
CountDownLatch countDownLatch = new CountDownLatch(threadNum);
for (List<Map> group : groupList) {
executorService.execute(()->{
try {
for (Map map : group) {
// update the data in mongodb
}
} catch (Exception e) {
e.printStackTrace();
} finally {
// let counter minus one
countDownLatch.countDown();
}
});
}
try {
// main thread donnot execute until all child threads finish
countDownLatch.await();
} catch (Exception e) {
e.printStackTrace();
}
// remember to shutdown the threadPool
executorService.shutdown();
return ResponseData.success();
}
??那么在使用多線程異步更新的策略后，從當(dāng)初調(diào)用接口所需的大概時(shí)間為 30-40 min 下降到了 8-10 min，大大提高了執(zhí)行效率。
??需要注意的是，這里使用的
newFixedThreadPool創(chuàng)建線程池，它有一個(gè)缺陷就是，它的阻塞隊(duì)列默認(rèn)是一個(gè)無界隊(duì)列，默認(rèn)值為Integer.MAX_VALUE極有可能會(huì)造成 OOM 問題。因此，一般可以使用ThreadPoolExecutor來創(chuàng)建線程池，自己可以指定等待隊(duì)列中的線程個(gè)數(shù)，避免產(chǎn)生 OOM 問題。
public ResponseData updateHostDept() {
// ...
List<Map> hostMapList = mongoTemplate.find(query, Map.class, "host");
// split the hostMapList for the following multi-threads task
// return the number of logical CPUs
int processorsNum = Runtime.getRuntime().availableProcessors();
// set the threadNum as 2*(the number of logical CPUs) for handling IO Tasks,
// if Computing Tasks set the threadNum as (the number of logical CPUs) + 1
int threadNum = processorsNum * 2;
// the number of each group data
int eachGroupNum = hostMapList.size() / threadNum;
List<List<Map>> groupList = new ArrayList<>();
for (int i = 0; i < threadNum; i++) {
int start = i * eachGroupNum;
if (i == threadNum - 1) {
int end = mapList.size();
groupList.add(hostMapList.subList(start, end));
} else {
int end = (i+1) * eachGroupNum;
groupList.add(hostMapList.subList(start, end));
}
}
// update data by using multi-threads asynchronously
ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 8, 30L, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(100));
CountDownLatch countDownLatch = new CountDownLatch(threadNum);
for (List<Map> group : groupList) {
executor.execute(()->{
try {
for (Map map : group) {
// update the data in mongodb
}
} catch (Exception e) {
e.printStackTrace();
} finally {
// let counter minus one
countDownLatch.countDown();
}
});
}
try {
// main thread donnot execute until all child threads finish
countDownLatch.await();
} catch (Exception e) {
e.printStackTrace();
}
// remember to shutdown the threadPool
executor.shutdown();
return ResponseData.success();
}在上述的代碼中，核心線程數(shù)和最大線程數(shù)分別為 5 和 8，并沒有設(shè)置的很大的值，因?yàn)槿绻绻O(shè)置的很大，線程間頻繁的上下文切換也會(huì)增加時(shí)間消耗，反而不能最大程度上發(fā)揮多線程的優(yōu)勢。至于如何選擇合適的參數(shù)，需要根據(jù)機(jī)器的參數(shù)以及任務(wù)的類型綜合考慮決定。
??最后補(bǔ)充一點(diǎn)，如果想要通過非編碼的方式獲取機(jī)器的 CPU 線程個(gè)數(shù)也很簡單，windows 系統(tǒng)通過任務(wù)管理器，選擇 “性能”，便可以查看 CPU 線程個(gè)數(shù)的情況，如下圖所示：
??從上圖可以看到，我的機(jī)器中內(nèi)核是八個(gè) CPU，但是通過超線程技術(shù)一個(gè)物理的 CPU 核心可以模擬成兩個(gè)邏輯 CPU 線程，因此我的機(jī)器是支持8核16線程的。
到此這篇關(guān)于Java 多線程并發(fā)編程提高數(shù)據(jù)處理效率的文章就介紹到這了,更多相關(guān)Java 多線程提高數(shù)據(jù)處理效率內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
這篇文章主要介紹了Java性能的十一個(gè)用法,需要的朋友可以參考下2014-10-10
Spring中的@Transactional事務(wù)失效場景解讀
這篇文章主要介紹了Spring中的@Transactional事務(wù)失效場景解讀,如果Transactional注解應(yīng)用在非public 修飾的方法上,Transactional將會(huì)失效此方法會(huì)檢查目標(biāo)方法的修飾符是否為 public,不是 public則不會(huì)獲取@Transactional 的屬性配置信息,需要的朋友可以參考下2023-12-12
這篇文章主要介紹了深入理解Java中的WeakHashMap,WeakHashMap從名字可以得知主要和Map有關(guān),不過還有一個(gè)Weak,我們就更能自然而然的想到這里面還牽扯到一種弱引用結(jié)構(gòu),因此想要徹底搞懂,我們還需要知道四種引用,需要的朋友可以參考下2023-09-09
java實(shí)現(xiàn)簡單的客戶信息管理系統(tǒng)
這篇文章主要為大家詳細(xì)介紹了java實(shí)現(xiàn)簡單的客戶信息管理系統(tǒng)，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2022-06-06
詳解spring boot應(yīng)用啟動(dòng)原理分析
這篇文章主要介紹了詳解spring boot應(yīng)用啟動(dòng)原理分析，小編覺得挺不錯(cuò)的，現(xiàn)在分享給大家，也給大家做個(gè)參考。一起跟隨小編過來看看吧2018-06-06
這篇文章主要介紹了java文件操作中的java寫文件示例,需要的朋友可以參考下2014-03-03
這篇文章主要介紹了Java多線程的死鎖，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2021-10-10
解決IDEA上循環(huán)依賴報(bào)錯(cuò)問題Error:java: Annotation processing&n
這篇文章主要介紹了解決IDEA上循環(huán)依賴報(bào)錯(cuò)問題Error:java: Annotation processing is not supported for module cycles,具有很好的參考價(jià)值,希望對(duì)大家有所幫助2023-10-10
這篇文章主要為大家詳細(xì)介紹了Spring實(shí)現(xiàn)郵件發(fā)送功能，簡單的發(fā)送郵件工具JavaMailSender使用，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2017-11-11
淺談Mybatis Plus的BaseMapper的方法是如何注入的
我們?cè)谟玫臅r(shí)候經(jīng)常就是生產(chǎn)自定義的Mapper繼承自BaseMapper，那么BaseMapper怎么被注入到mybatis里的，本文就詳細(xì)的介紹一下，感興趣的可以了解一下2021-09-09