第一步：切割数据的方法
public static List<List<Integer>> getProcessData(Integer batchSize) {
List<Integer> originData = new ArrayList<>();
for (int i = 0; i < 100; i++) {
originData.add(i);
}
List<List<Integer>> processData = ListUtils.partition(originData, batchSize);
System.out.println(processData.size());
return processData;
}
1.入参是 batchSize，我们想要切割的大小（这里设置的是5）
2.然后for循环添加入我们的List数组originData中
3.ListUtils.partition是分割方法，这里既按照batchSize吧0-99分为固定的份数丢入我们的List<List> 中如上所示，拆成了二十份数据
ListUtils.partition的源码解释，意思就是把abcde的list数组拆成List<List>的格式
第二步： 单线程转换数据的方法
public static List<String> processOneThread(List<Integer> req) throws InterruptedException {
List<String> output = new ArrayList<>();
for (Integer value : req) {
output.add(String.valueOf(value+ 1));
}
Thread.sleep(200);
return output;
}
单线程数据处理,把List类型转换为List类型
1.入参是 List req
2.通过for循环的方式以及String.valueof把int类型转换为String类型，这里value+1的原因是切割数据的方法是从0开始的，下标我们就要+1让他有数据。
3.Thread.sleep睡眠的原因是为了更好的区分度
4.最终出参的形式是List
第三步：多线程调用
public static List<String> processByMultiThread(Integer batchSize) throws ExecutionException, InterruptedException {
List<String> output = new ArrayList<>();
// 获取分批数据
List<List<Integer>> batchProcessData = getProcessData(batchSize);
// 启动线程
List<FutureTask<List<String>>> futureTaskList = new ArrayList<>();
for (List<Integer> processData : batchProcessData) {
Callable<List<String>> callable = () -> processOneThread(processData);
FutureTask<List<String>> futureTask = new FutureTask<>(callable);
new Thread(futureTask).start();
// 启动线程
futureTaskList.add(futureTask);
}
// 获取线程返回的数据
for (FutureTask futureTask : futureTaskList) {
List<String> processData = (List<String>) futureTask.get();
output.addAll(processData);
}
return output;
}
主要解析代码
List<FutureTask<List<String>>> futureTaskList = new ArrayList<>();
for (List<Integer> processData : batchProcessData) {
Callable<List<String>> callable = () -> processOneThread(processData);
FutureTask<List<String>> futureTask = new FutureTask<>(callable);
new Thread(futureTask).start();
// 启动线程
futureTaskList.add(futureTask);
}
List<FutureTask<List<String>>> futureTaskList = new ArrayList<>();
1.最内层 就是我们通过processOneThread方法把List类型转换为List类型
2.第二层就是调用分批次处理方法把我们的数据切分成5个为一个单位的批次，例如这里是200个切分成20次，即对应的
FutureTask<List<String>> futureTask = new FutureTask<>(callable);
用futureTast接收,FutureTask是一个包装器，它通过接受Callable来创建，它同时实现了Future和Runnable接口,由FutureTask创建一个Thread对象。
3. 每一个批次就启动一个线程去处理
new Thread(futureTask).start();
- 再把处理把每一个批次的list丢进我们进行收集的futureTaskList也就是最外层的list中
futureTaskList.add(futureTask);
这里说到底还是用单线程去处理整个的问题，只是分批次了，即并行的思想
如何获取返回的数据
// 获取线程返回的数据
for (FutureTask futureTask : futureTaskList) {
List<String> processData = (List<String>) futureTask.get();
output.addAll(processData);
}
很明显的是通过futureTask.get()方法去获取我们要返回的数据
但是ctrl+鼠标左键点进这个方法会发现实现的是我们futureTast下的get方法
1.先查看当前线程的状态值
2.对状态值进行判定，是否创建线程
3.如果状态值是就绪态以上就进入awaitDown方法
状态值：
awaitDown方法
第三步替换，用线程池去处理
public static List<String> processByPool(Integer batchSize) throws ExecutionException, InterruptedException {
List<String> output = new ArrayList<>();
List<List<Integer>> batchProcessData = getProcessData(batchSize);
// 1. 使用线程池
ExecutorService executorService = new ThreadPoolExecutor(3, 5, 60,
TimeUnit.SECONDS,
new LinkedBlockingDeque<Runnable>(10), Executors.defaultThreadFactory(),
new ThreadPoolExecutor.CallerRunsPolicy());
// 2. 创建执行任务
List<Future<List<String>>> futureList = new ArrayList<>();
for (List<Integer> processData : batchProcessData) {
Callable<List<String>> callable = () -> processOneThread(processData);
futureList.add(executorService.submit(callable));
}
for (Future futureTask : futureList) {
List<String> processData = (List<String>) futureTask.get();
output.addAll(processData);
}
return output;
}
不同之处：
ExecutorService executorService = new ThreadPoolExecutor(3, 5, 60,
TimeUnit.SECONDS,
new LinkedBlockingDeque<Runnable>(10), Executors.defaultThreadFactory(),
new ThreadPoolExecutor.CallerRunsPolicy());
// 2. 创建执行任务
List<Future<List<String>>> futureList = new ArrayList<>();
for (List<Integer> processData : batchProcessData) {
Callable<List<String>> callable = () -> processOneThread(processData);
futureList.add(executorService.submit(callable));
}
List<Future<List<String>>> futureList = new ArrayList<>();
for (List<Integer> processData : batchProcessData) {
Callable<List<String>> callable = () -> processOneThread(processData);
FutureTask<List<String>> futureTask = new FutureTask<>(callable);
new Thread(futureTask).start();
// 启动线程
futureTaskList.add(futureTask);
}
- 第一个不同就是线程池开始的时候会构造一个线程池，丢入一些核心参数
- 第二个不同就是在把每个线程执行的结果futureTask通过内部的方法丢入我们的futureList收集中
即
futureList.add(executorService.submit(callable));
我们点进这个submit方法会发现进入了execytorService这个接口，但是实际的方法肯定是ThreadPoolExecutor这个实现类下的，所以我们就进入这个实现类找submit方法
最终在ThreadPoolExecutor的父类AbstractExecutorService中找到了这个submit方法
这个submit方法最终就是要把我们传入的Callable返回，即内部处理了我们多线程处理方法的
FutureTask<List<String>> futureTask = new FutureTask<>(callable);
