一、場景描述

例如: 需要指定3個線程來處理1000條數(shù)據(jù), 其中線程數(shù)、數(shù)據(jù)量是可變的。

二、 思路

1. 針對數(shù)據(jù)量進行分片， 片的數(shù)量即線程數(shù)。
2. 控制線程數(shù)可通過CountDownLatch來處理。

三、代碼

1. 程序執(zhí)行類

/**
 * @ClassName : LimitThreadNumApp
 * @Description : 限定線程的執(zhí)行數(shù)，去執(zhí)行大數(shù)據(jù)
 * @Author : hack2012
 * @Date: 2020-12-16 22:53
 */
public class LimitThreadNumApp {
    Logger log = LoggerFactory.getLogger(getClass());

    static final  ExecutorService executorService = Executors.newFixedThreadPool(3);

    @Test
    public void testProcess() throws Exception{
        //數(shù)據(jù)量
        int size = 100;
        //定義執(zhí)行任務的線程數(shù)
        int threadNum = 3;

        process(threadNum, size);
    }

    public  void process(int threadNum, int size) throws InterruptedException {

        List<TaskModel> datas = initData(size);

        //計算每個分片的大小
        int shardSize = datas.size() / threadNum + 1;

        CountDownLatch latch = new CountDownLatch(threadNum);
        int cnt = 0;
        for (int i = 0; i < threadNum; i++) {
            List<TaskModel> processDatas = shardTask(datas, i, shardSize);
            cnt += processDatas.size();
            executorService.submit(new TaskCallable(processDatas, latch));
        }
        latch.await();

        log.info("預期數(shù)據(jù)量:{};實際執(zhí)行數(shù)據(jù)量:{}", size, cnt);
    }

    /** 任務分片 */
    private List<TaskModel> shardTask(List<TaskModel> datas, int tnum, int size){
        int start = tnum * size;
        int end = (tnum + 1) * size;
        if (end > datas.size()) {
            end = datas.size();
        }
        return datas.subList(start, end);
    }

    private  List<TaskModel> initData(int size){
        List<TaskModel> datas = Lists.newArrayList();
        for (int i = 1; i <= size; i++) {
            TaskModel data = new TaskModel();
            data.setId(String.valueOf(i));
            datas.add(data);
        }
        return datas;
    }
}

2. 定義任務模型

/**
 * @ClassName : TaskModel
 * @Description : 數(shù)據(jù)模型
 * @Author : hack2012
 * @Date: 2020-12-16 22:55
 */
public class TaskModel {

    private String id ;


    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }
}

3. 定義任務執(zhí)行類

/**
 * @ClassName : TaskCallable
 * @Description : 任務執(zhí)行器
 * @Author : hack2012
 * @Date: 2020-12-16 22:58
 */
public class TaskCallable implements Callable<Boolean> {
    Logger log = LoggerFactory.getLogger(getClass());

    List<TaskModel> datas;
    CountDownLatch latch;

    TaskCallable(List<TaskModel> datas, CountDownLatch latch) {
        this.datas = datas;
        this.latch = latch;
    }

    @Override
    public Boolean call() throws Exception {
        try {
            int cnt = 0;
            for (TaskModel data : datas) {
                cnt++;
            }

            log.info("[{}]-線程執(zhí)行數(shù)量:{}", Thread.currentThread().getName(), cnt);

            latch.countDown();
            return true;
        } catch (Exception e) {
            log.error(Thread.currentThread().getName() + "線程執(zhí)行失敗:{}", e.getMessage());
        }
        return false;
    }
}

四、測試結果

1. 線程3， 數(shù)量 98

[2020-12-1623:36:55下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-1]-線程執(zhí)行數(shù)量:33
-[2020-12-1623:36:55下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-3]-線程執(zhí)行數(shù)量:32
-[2020-12-1623:36:55下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-2]-線程執(zhí)行數(shù)量:33
-[2020-12-1623:36:55下午]:INFOthread.LimitThreadNumApp.process(LimitThreadNumApp.java:44)預期數(shù)據(jù)量:98;實際執(zhí)行數(shù)據(jù)量:98

1. 線程3， 數(shù)量 99

[2020-12-1623:37:40下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-3]-線程執(zhí)行數(shù)量:31
-[2020-12-1623:37:40下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-1]-線程執(zhí)行數(shù)量:34
-[2020-12-1623:37:40下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-2]-線程執(zhí)行數(shù)量:34
-[2020-12-1623:37:40下午]:INFOthread.LimitThreadNumApp.process(LimitThreadNumApp.java:44)預期數(shù)據(jù)量:99;實際執(zhí)行數(shù)據(jù)量:99

1. 線程3， 數(shù)量 100

[2020-12-1623:38:18下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-1]-線程執(zhí)行數(shù)量:34
-[2020-12-1623:38:18下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-2]-線程執(zhí)行數(shù)量:34
-[2020-12-1623:38:18下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-3]-線程執(zhí)行數(shù)量:32
-[2020-12-1623:38:18下午]:INFOthread.LimitThreadNumApp.process(LimitThreadNumApp.java:44)預期數(shù)據(jù)量:100;實際執(zhí)行數(shù)據(jù)量:100