多線程快速處理 List 集合，你學(xué)會了嗎？

來自：CSDN，作者：qililong88

鏈接：https://blog.csdn.net/qililong88/article/details/114320641

有一個(gè)大List集合，遍歷進(jìn)行一些耗時(shí)操作，不能達(dá)到性能要求，查詢?nèi)罩?，單個(gè)任務(wù)雖然有不少數(shù)據(jù)庫和第三方API請求，比較耗時(shí)，但返回效率尚可，所以優(yōu)先采用多線程方式進(jìn)行處理并行請求數(shù)據(jù)庫和第三方API，因?yàn)樘幚硗赀€要對list所屬的數(shù)據(jù)進(jìn)行操作，所以，線程池多線程處理要等待全部處理完。

相關(guān)的代碼如下：

@Test
public void testTB()
{
  List < String > list = new ArrayList < > ();
  for(int i = 0; i < 900; i++)
  {
    list.add("a");
  }
  ExecutorService touchWorker = Executors.newFixedThreadPool(4, new ThreadFactoryBuilder().setNameFormat("touch-send-worker-%d").build());
  int size = list.size();
  if(size > 100)
  {
    int batch = size % 100 == 0 ? size / 100 : size / 100 + 1;
    for(int j = 0; j < batch; j++)
    {
      int end = (j + 1) * 100;
      if(end > size)
      {
        end = size;
      }
      List < String > subList = list.subList(j * 100, end);
      touchWorker.execute(() - > sleepTest(subList));
    }
    touchWorker.shutdown();
    while(true)
    {
      if(touchWorker.isTerminated())
      {
        break;
      }
    }
  }
  else
  {
    sleepTest(list);
  }
}
private void sleepTest(List < String > subList)
{
  for(String i: subList)
  {
    try
    {
      //耗時(shí)操作
      System.out.println("######" + i + "######" + Thread.currentThread().getName());
      //                Thread.sleep(1000*30);
    }
    catch(Exception e)
    {
      e.printStackTrace();
    }
  }
}

void shutdown()

啟動一次順序關(guān)閉，執(zhí)行以前提交的任務(wù)，但不接受新任務(wù)。若已經(jīng)關(guān)閉，則調(diào)用沒有其他作用。

拋出：SecurityException - 如果安全管理器存在并且關(guān)閉，此 ExecutorService 可能操作某些不允許調(diào)用者修改的線程（因?yàn)樗鼪]有保持RuntimePermission("modifyThread")），或者安全管理器的 checkAccess 方法拒絕訪問。

boolean isTerminated()

若關(guān)閉后所有任務(wù)都已完成，則返回true。注意除非首先調(diào)用shutdown或shutdownNow，否則isTerminated永不為true。返回：若關(guān)閉后所有任務(wù)都已完成，則返回true。

當(dāng)然還有一種方法，是之前寫的，方法比上邊的臃腫了，不過會獲取返回結(jié)果進(jìn)行處理：邏輯是獲取所有頁面的List，多線程遍歷List后，將所有頁面的違規(guī)詞查出發(fā)送郵件，代碼：

/**
 *  落地頁違規(guī)詞排查(多線程)。
 * @return
 */
@Test
public  void getViolationsLandpageByThreadPool() {
    try {
        landPageService.getViolationsLandpageByThreadPool("1年");
    } catch (Exception e) {
        e.printStackTrace();
    }
}

// 開始時(shí)間
long start = System.currentTimeMillis();
/*List<LandPage> list = new ArrayList<LandPage>();
for (int i = 1; i <= 3000; i++) {
    list.add(i + "");
}*/

List<LandPage> list = landPageDao.getPageIdAndPath();
// 初始化敏感詞庫對象
SensitiveWordInit sensitiveWordInit = new SensitiveWordInit();
// 從數(shù)據(jù)庫中獲取敏感詞對象集合（目前先放在資源文件中，等以后比較多或者需要界面操作時(shí)再用數(shù)據(jù)庫）
// 構(gòu)建敏感詞庫
Map sensitiveWordMap = sensitiveWordInit.initKeyWord(vioKey);
// 傳入SensitivewordEngine類中的敏感詞庫
SensitivewordEngine.sensitiveWordMap = sensitiveWordMap;
// 每500條數(shù)據(jù)開啟一條線程
int threadSize = 11000;
// 總數(shù)據(jù)條數(shù)
int dataSize = list.size();
// 線程數(shù)
int threadNum = dataSize / threadSize + 1;
// 定義標(biāo)記,過濾threadNum為整數(shù)
boolean special = dataSize % threadSize == 0;

/*list.parallelStream().forEach(url ->{});*/
// 創(chuàng)建一個(gè)線程池
ExecutorService exec = Executors.newFixedThreadPool(threadNum);
// 定義一個(gè)任務(wù)集合
List<Callable<List<LandPage>>> tasks = new ArrayList<Callable<List<LandPage>>>();
Callable<List<LandPage>> task = null;
List<LandPage> 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));
    }
    // System.out.println("第" + (i + 1) + "組：" + cutList.toString());
    final List<LandPage> listStr = cutList;
    task = new Callable<List<LandPage>>() {

        @Override
        public List<LandPage> call() throws Exception {
//                    System.out.println(Thread.currentThread().getName() + "線程：" + listStr.get(0).getPageId());
             List<LandPage> result = new ArrayList<LandPage>();
             for (LandPage landPage : listStr) {
                 Long pageId = landPage.getPageId();
                 String path = landPage.getPath();
                 Integer version = landPage.getVersion();
                 String pageUrl = landPage.getPageUrl();
                 String actualUser = landPage.getActualUser();
                 Integer pageType = landPage.getPageType();
                 if (StringUtils.isNotBlank(path)) {
                     // 調(diào)用第一個(gè)方法，獲取html字符串
                     String html = httpRequest(path);
                     // 調(diào)用第二個(gè)方法，獲取包含的違規(guī)詞
                     if(StringUtils.isNotBlank(html)){
                         html = html.replaceAll("<!--(.*?)-->","");
//      String buffer = htmlFiter2(html);
                        Set<String> bufferSet = SensitivewordEngine.getSensitiveWord(html, 1);// 得到敏感詞有哪些，傳入2表示獲取所有敏感詞//sensitiveWordFiltering(html);
                        /*String[] word = {"備案","錯過將延誤","僅需1980元"};
                        for(int i=0 ;i<word.length;i++){
                            if(html.contains(word[i])){
                                bufferSet.add(word[i]);
                            }
                        }*/
                        String[] word = {
                            "一年","1年學(xué)完","一年學(xué)完","1年內(nèi)學(xué)完","一年內(nèi)學(xué)完"
                        };
                        for(int i=0 ;i<word.length;i++){
                            if(html.contains(word[i])){
                                bufferSet.add(word[i]);
                            }
                        }
                        if (null!=bufferSet&&bufferSet.size()>0) {

                            String sensitiveWord = bufferSet == null ? null : bufferSet.toString();
                            if ("[]".equals(sensitiveWord)){
                                sensitiveWord = "";
                            }
                            LandPage page = new LandPage();
                            page.setPageId(pageId);
                            page.setPath(path);
                            page.setVersion(version);
                            page.setDescription(sensitiveWord);
                            page.setPageUrl(pageUrl);
                            page.setActualUser(actualUser);
                            page.setPageType(pageType);
                            result.add(page);
                            System.out.println(pageUrl);
                        }
                    }
                }
            }
            return (List<LandPage>) result;
        }
    };
    // 這里提交的任務(wù)容器列表和返回的Future列表存在順序?qū)?yīng)的關(guān)系
    tasks.add(task);
}

List<Future<List<LandPage>>> results = exec.invokeAll(tasks);
List<LandPage> result = new ArrayList<LandPage>();
for (Future<List<LandPage>> future : results) {
    result.addAll(future.get());
}

// 關(guān)閉線程池
exec.shutdown();
System.out.println("線程任務(wù)執(zhí)行結(jié)束");
System.err.println("執(zhí)行任務(wù)消耗了 ：" + (System.currentTimeMillis() - start) + "毫秒");

System.out.println("共有###########"+list.size() );

result就是需要發(fā)送郵件的相關(guān)結(jié)果了