97.java 中通过动态指定线程来执行任务列表

一、场景描述

例如: 需要指定3个线程来处理1000条数据, 其中线程数、数据量是可变的。

二、 思路

1. 针对数据量进行分片， 片的数量即线程数。
2. 控制线程数可通过CountDownLatch来处理。

三、代码

1. 程序执行类

/**
 * @ClassName : LimitThreadNumApp
 * @Description : 限定线程的执行数，去执行大数据
 * @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{
        //数据量
        int size = 100;
        //定义执行任务的线程数
        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("预期数据量:{};实际执行数据量:{}", 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 : 数据模型
 * @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. 定义任务执行类

/**
 * @ClassName : TaskCallable
 * @Description : 任务执行器
 * @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("[{}]-线程执行数量:{}", Thread.currentThread().getName(), cnt);

            latch.countDown();
            return true;
        } catch (Exception e) {
            log.error(Thread.currentThread().getName() + "线程执行失败:{}", e.getMessage());
        }
        return false;
    }
}

四、测试结果

1. 线程3， 数量 98

[2020-12-1623:36:55下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-1]-线程执行数量:33
-[2020-12-1623:36:55下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-3]-线程执行数量:32
-[2020-12-1623:36:55下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-2]-线程执行数量:33
-[2020-12-1623:36:55下午]:INFOthread.LimitThreadNumApp.process(LimitThreadNumApp.java:44)预期数据量:98;实际执行数据量:98

1. 线程3， 数量 99

[2020-12-1623:37:40下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-3]-线程执行数量:31
-[2020-12-1623:37:40下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-1]-线程执行数量:34
-[2020-12-1623:37:40下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-2]-线程执行数量:34
-[2020-12-1623:37:40下午]:INFOthread.LimitThreadNumApp.process(LimitThreadNumApp.java:44)预期数据量:99;实际执行数据量:99

1. 线程3， 数量 100

[2020-12-1623:38:18下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-1]-线程执行数量:34
-[2020-12-1623:38:18下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-2]-线程执行数量:34
-[2020-12-1623:38:18下午]:INFOthread.TaskCallable.call(TaskCallable.java:35)[pool-1-thread-3]-线程执行数量:32
-[2020-12-1623:38:18下午]:INFOthread.LimitThreadNumApp.process(LimitThreadNumApp.java:44)预期数据量:100;实际执行数据量:100