Java多線程處理List問題

更新時間：2023年09月21日 16:25:44 作者：JonTang

這篇文章主要介紹了Java多線程處理List問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教

Java多線程處理List

項目場景

調(diào)用第三方提供的接口去獲取 List 中用戶的組信息。

問題描述

需要拿用戶的 id 去調(diào)用第三方接口，成功調(diào)用一次需要 0.3s 左右，當(dāng)有 1000 個用戶時，就需要花費 0.3 * 1000s = 5min，頁面就會一直加載那么久。

之前是通過 for 循環(huán) list 去調(diào)用接口的，

代碼如下：

// 當(dāng) list 長度為 1000時，則需要循環(huán) 1000次
for(User user : list) {
    loadUserGroups(user);
}

解決方案

通過多線程的方式去處理，話不多說直接上代碼：

// 定義一個線程池
private static final ExecutorService loadUserGroupsExecutor = Executors.newFixedThreadPool(20);
public Map<String, List<UserGroup>> loadAllUserGroups() {
        Map<String, List<UserGroup>> userGroups = new ConcurrentHashMap<>();
        List<User> users = listUsers();
        int size = users.size();       
        long startTime = System.currentTimeMillis();        
        if (size > 200) {            
            List<List<User>> partition = Lists.partition(users, 200);            
            List<CompletableFuture> results = new ArrayList<>();            
            for (List<User> subList : partition) {                
                CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {                    
                loadUserGroups(userGroups, subList);                    
                return "";                
                }, loadUserGroupsExecutor);                
            	    results.add(future);            
                }            
            CompletableFuture.allOf(results.toArray(results.toArray(new CompletableFuture[partition.size()]))).join();        
        } else {            
        	loadUserGroups(userGroups, users);        
        }
    	log.info("loadAllUserGroups cost {}", System.currentTimeMillis() - startTime);        
     return userGroups;
}

Java多線程分段處理List集合

場景：

大數(shù)據(jù)List集合，需要對List集合中的數(shù)據(jù)同標(biāo)準(zhǔn)庫中數(shù)據(jù)進(jìn)行對比,生成新增，更新，取消數(shù)據(jù)

解決方案

List集合分段
動態(tài)創(chuàng)建線程池newFixedThreadPool
將對比操作在多線程中實現(xiàn)

public static void main(String[] args) throws Exception {
?? ?// 開始時間
?? ?long start = System.currentTimeMillis();
?? ?List<String> list = new ArrayList<String>();
?? ?for (int i = 1; i <= 3000; i++) {
?? ??? ?list.add(i + "");
?? ?}
? ? /*動態(tài)線程數(shù)方式*/
?? ?// 每500條數(shù)據(jù)開啟一條線程
?? ?int threadSize = 500;
?? ?// 總數(shù)據(jù)條數(shù)
?? ?int dataSize = list.size();
?? ?// 線程數(shù)，動態(tài)生成
?? ?int threadNum = dataSize / threadSize + 1;
? ? /*固定線程數(shù)方式
?? ? ? ?// 線程數(shù)
?? ? ? ?int threadNum = 6;
?? ? ? ?// 總數(shù)據(jù)條數(shù)
?? ? ? ?int dataSize = list.size();
?? ? ? ?// 每一條線程處理多少條數(shù)據(jù)
?? ? ? ?int threadSize = dataSize / (threadNum - 1);
? ? */
?? ?// 定義標(biāo)記,過濾threadNum為整數(shù)
?? ?boolean special = dataSize % threadSize == 0;
?? ?// 創(chuàng)建一個線程池
?? ?ExecutorService exec = Executors.newFixedThreadPool(threadNum);
?? ?// 定義一個任務(wù)集合
?? ?List<Callable<Integer>> tasks = new ArrayList<Callable<Integer>>();
?? ?Callable<Integer> task = null;
?? ?List<String> cutList = null;
?? ?// 確定每條線程的數(shù)據(jù)
?? ?for (int i = 0; i < threadNum; i++) {
?? ??? ?if (i == threadNum - 1) {
?? ??? ??? ?if (special) {
?? ??? ??? ??? ?break;
?? ??? ??? ?}
?? ??? ??? ?cutList = list.subList(threadSize * i, dataSize);
?? ??? ?} else {
?? ??? ??? ?cutList = list.subList(threadSize * i, threadSize * (i + 1));
?? ??? ?}
?? ??? ?final List<String> listStr = cutList;
?? ??? ?task = new Callable<Integer>() {
?? ??? ??? ?@Override
?? ??? ??? ?public Integer call() throws Exception {
?? ??? ??? ??? ?//業(yè)務(wù)邏輯，循環(huán)處理分段后的list
?? ??? ??? ??? ?System.out.println(Thread.currentThread().getName() + "線程：" + listStr);
?? ??? ??? ??? ?//......
?? ??? ??? ??? ?return 1;
?? ??? ??? ?}
?? ??? ?};
?? ??? ?// 這里提交的任務(wù)容器列表和返回的Future列表存在順序?qū)?yīng)的關(guān)系
?? ??? ?tasks.add(task);
?? ?}
?? ?exec.invokeAll(tasks);
?? ?// 關(guān)閉線程池
?? ?exec.shutdown();
?? ?System.out.println("線程任務(wù)執(zhí)行結(jié)束");
?? ?System.out.println("執(zhí)行任務(wù)消耗了 ：" + (System.currentTimeMillis() - start) + "毫秒");
}