package com.ithzk.threadpool;

/**
 * 用作返回 執(zhí)行的數(shù)量的
 * @author hzk
 * @date 2018/3/29
 */
public class LogNumVo {
  private static final long serialVersionUID = -5541722936350755569L;
  private Integer dataNum;
  private Integer successNum;
  private Integer waitNum;

  public Integer getDataNum() {
    return dataNum;
  }
  public void setDataNum(Integer dataNum) {
    this.dataNum = dataNum;
  }
  public Integer getSuccessNum() {
    return successNum;
  }
  public void setSuccessNum(Integer successNum) {
    this.successNum = successNum;
  }
  public Integer getWaitNum() {
    return waitNum;
  }
  public void setWaitNum(Integer waitNum) {
    this.waitNum = waitNum;
  }
}
 

package com.ithzk.threadpool;

/**
 * @author hzk
 * @date 2018/3/29
 */
public class DealObject {

  private Integer identifyId;

  private String data;

  public DealObject(Integer identifyId, String data) {
    this.identifyId = identifyId;
    this.data = data;
  }

  public DealObject() {
  }

  public Integer getIdentifyId() {
    return identifyId;
  }

  public void setIdentifyId(Integer identifyId) {
    this.identifyId = identifyId;
  }

  public String getData() {
    return data;
  }

  public void setData(String data) {
    this.data = data;
  }
}

package com.ithzk.threadpool;

import java.util.Collection;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;

/**
 * @author hzk
 * @date 2018/3/29
 */
public class AbstractCalculateThread<T> implements Callable<String> {

  protected Collection<T> insertList;

  protected CountDownLatch countd;

  protected String threadCode;

  protected String batchNumber;

  public Collection<T> getInsertList() {
    return insertList;
  }

  public void setInsertList(Collection<T> insertList) {
    this.insertList = insertList;
  }

  public CountDownLatch getCountd() {
    return countd;
  }

  public void setCountd(CountDownLatch countd) {
    this.countd = countd;
  }

  public String getThreadCode() {
    return threadCode;
  }

  public void setThreadCode(String threadCode) {
    this.threadCode = threadCode;
  }

  public String getBatchNumber() {
    return batchNumber;
  }

  public void setBatchNumber(String batchNumber) {
    this.batchNumber = batchNumber;
  }


  public AbstractCalculateThread() {
    super();
  }

  public AbstractCalculateThread(Collection<T> insertList, CountDownLatch countd, String threadCode,String batchNumber) {
    super();
    this.insertList = insertList;
    this.countd = countd;
    this.threadCode = threadCode;
    this.batchNumber = batchNumber;
  }

  public String call() throws Exception {
    return null;
  }
}

package com.ithzk.threadpool;

import java.util.Collection;
import java.util.concurrent.CountDownLatch;

/**
 * @author hzk
 * @date 2018/3/29
 */
public class CalculateDealThread extends AbstractCalculateThread<DealObject> {

  private ExecutorPool executorPool = SpringContextUtil.getBean(ExecutorPool.class);

  @Override
  public String call() throws Exception {
    try {
      System.out.println("========開(kāi)始跑線(xiàn)程【"+threadCode+"】");
      return executorPool.syncBatchDealObject(insertList,batchNumber);
    } catch (Exception e) {
      e.printStackTrace();
      System.out.println("========開(kāi)始跑線(xiàn)程【"+threadCode+"】:"+e.getMessage());
    }finally {
      countd.countDown();
    }
    return null;
  }

  public CalculateDealThread() {
    super();
  }

  public CalculateDealThread(Collection<DealObject> insertList, CountDownLatch countd, String threadCode,String batchNumber) {
    super(insertList, countd, threadCode, batchNumber);
  }

}

package com.ithzk.threadpool;

import java.util.*;
import java.util.concurrent.*;

/**
 * @author hzk
 * @date 2018/3/29
 */
public class ExecutorPool {

  /**
   * 模擬需要處理數(shù)據(jù)的大小
   */
  private static final int ARRAY_COUNT = 50000;
  /**
   * 開(kāi)啟多線(xiàn)程處理的條件
   */
  private static final int MULTI_THREAD_STARTCOUNT = 10000;
  /**
   * 批量處理的大小
   */
  private static final int BATCH_DEAL_SIZE = 100;
  /**
   * 每次開(kāi)啟線(xiàn)程數(shù)量
   */
  public static final int THREAD_POOL_NUM=10;

  public static void main(String[] args){
    testExecutorPool();
  }

  public static void testExecutorPool(){
    ArrayList<DealObject> dealObjects = new ArrayList<DealObject>();
    for (int i = 0;i<ARRAY_COUNT;i++){
      DealObject dealObject = new DealObject(i,"data_"+i);
      dealObjects.add(dealObject);
      System.out.println("Data add success current:"+i);
    }
    int size = dealObjects.size();
    int successNum = 0;
    int waitNum = 0;
    System.out.println("需要處理的數(shù)據(jù)數(shù)據(jù)量為:"+size);
    // 判斷數(shù)據(jù)是否大于10000 如果大于則開(kāi)啟線(xiàn)程池 跑數(shù)據(jù)
    if (size > MULTI_THREAD_STARTCOUNT) {
      try {
        System.out.println("===================dataNum > 1000 | Multiple Thread Run=======================");
        // 每次新增處理的條數(shù)
        int batchInsertSize = BATCH_DEAL_SIZE;
        // 定義保存的線(xiàn)程池
        ExecutorService executorInsert = Executors.newFixedThreadPool(THREAD_POOL_NUM);
        // 定義保存過(guò)程中返回的線(xiàn)程執(zhí)行返回參數(shù)
        List<Future<String>> futureListIsert = new ArrayList<Future<String>>();
        // 線(xiàn)程 修改list
        List<Map<Integer, DealObject>> listDealObjects = new ArrayList<Map<Integer, DealObject>>();
        List<Map<Integer, DealObject>> listLiveSyncLogInsert = pointDateClassify(dealObjects, batchInsertSize, listDealObjects);
        if (null != listLiveSyncLogInsert && !listDealObjects.isEmpty()) {
          System.out.println("===================切割后的大小:"+listLiveSyncLogInsert.size()+"=======================");
          //配合使用CountDownLatch為了保證在執(zhí)行完所有子程序之后再執(zhí)行主程序
          CountDownLatch countd = new CountDownLatch(listLiveSyncLogInsert.size());
          for (int j = 0; j < listLiveSyncLogInsert.size(); j++) {
            Map<Integer, DealObject> insert = listLiveSyncLogInsert.get(j);
            Future<String> future = executorInsert.submit(new CalculateDealThread(insert.values(), countd,"executor_pool_test_thread", null));
            futureListIsert.add(future);
          }
        }
        // 等待線(xiàn)程執(zhí)行完成
        executorInsert.shutdown();
        for (Future<String> future : futureListIsert) {
          String json = future.get();
          if (null != json && !"".equals(json)) {
            將返回的json格式數(shù)據(jù)轉(zhuǎn)換為實(shí)體類(lèi) 進(jìn)行業(yè)務(wù)記錄
            LogNumVo logNumVo = JSON.toJavaObject(JSON.parseObject(json),LogNumVo.class);
            successNum += logNumVo.getSuccessNum();
            waitNum += logNumVo.getWaitNum();
          }
        }
      } catch (InterruptedException e) {
        e.printStackTrace();
      } catch (ExecutionException e) {
        e.printStackTrace();
      }

    }
  }

  /**
   * 拆分線(xiàn)程數(shù)
   * 假設(shè)集合中有50000個(gè)元素 則按照100個(gè)一組切分 可切分為500組
   * 即每個(gè)線(xiàn)程一次處理一組(100個(gè)元素)
   *
   * @author
   * @param lPostUploadIntegralList
   * @param batchInsertSize
   * @param listPostUploadIsert
   */
  @SuppressWarnings("all")
  public static List<Map<Integer, DealObject>> pointDateClassify(List<DealObject> lPostUploadIntegralList,int batchInsertSize, List<Map<Integer, DealObject>> listJSONObjectUpdate) {
    List<Map<Integer, DealObject>> listLiveSyncLogInsert = new Vector<Map<Integer, DealObject>>();
    // 新增數(shù)據(jù)list
    List<DealObject> integralListInsert = lPostUploadIntegralList;

    System.out.println("============integralListInsert.size()=====:" + integralListInsert.size());
    // 拆分?jǐn)?shù)據(jù)(拆成多個(gè)List)
    int inserti = 0;
    if (integralListInsert != null && integralListInsert.size() > 0) {
      ConcurrentHashMap<Integer, DealObject> integralListIns = null;
      for (int l = 0; l < integralListInsert.size(); l++) {
        if (integralListIns == null) {
          integralListIns = new ConcurrentHashMap<Integer, DealObject>();
        }
        integralListIns.put(integralListInsert.get(l).getIdentifyId(), integralListInsert.get(l));
        inserti++;
        if ((inserti % batchInsertSize) == 0) {
          listLiveSyncLogInsert.add(integralListIns);
          integralListIns = null;
        } else {
          // 最后100條或不足100條數(shù)據(jù)
          if ((l + 1) == integralListInsert.size()) {
            listLiveSyncLogInsert.add(integralListIns);
          }
        }
      }
    }
    System.out.println("=============listPostUploadInsert.size()====:" + listLiveSyncLogInsert.size());
    return listLiveSyncLogInsert;
  }

  /**
   * 多線(xiàn)程保存數(shù)據(jù)至數(shù)據(jù)庫(kù)
   */
  public String syncBatchDealObject(Collection<DealObject> insertList,String batchNumber) {
    int successNum = 0, waitNum = 0;
    Date currentDate = new Date(System.currentTimeMillis());
    for (DealObject dealObject : insertList) {
      try {
        int icount = syncDealObject(dealObject,currentDate);
        if(icount > 0){
          successNum ++;
        }else {
          waitNum ++;
        }
      } catch (Exception e) {
        e.printStackTrace();
        ++waitNum;
      }
    }
    LogNumVo logNum = new LogNumVo();
    logNum.setDataNum(0);
    logNum.setSuccessNum(successNum);
    logNum.setWaitNum(waitNum);
    // 將記錄實(shí)體類(lèi)轉(zhuǎn)為json格式反饋給線(xiàn)程池
    return JSON.toJSONString(logNum);
  }

  /**
   * 處理數(shù)據(jù)業(yè)務(wù)
   * @param dealObject
   * @param currentDate
   * @return
   */
  private int syncDealObject(DealObject dealObject,Date currentDate){
    int successNum = 0;
    //業(yè)務(wù)處理邏輯
    if(null != dealObject.getData()){
      successNum++;
    }
    return successNum;
  }
}