难点：如何获取子线程的执行结果并聚合
思路一
生产者-消费者 经典模型：
多个生产者负责生产(累加)作业
生产者将生产结果存入共享仓库中
消费者（主线程）从共享仓库中取出结果
/**
* 多线程计算累加数
*/
public class Accumulate {
public static void main(String[] args) {
Storage storage = new Storage();
// 为多个计算器对象创建线程
Thread calThread1 = new Thread(new Calculate(1, storage), "Thread-1");
Thread calThread2 = new Thread(new Calculate(2, storage), "Thread-2");
Thread calThread3 = new Thread(new Calculate(3, storage), "Thread-3");
Thread calThread4 = new Thread(new Calculate(4, storage), "Thread-4");
Thread calThread5 = new Thread(new Calculate(5, storage), "Thread-5");
Thread calThread6 = new Thread(new Calculate(6, storage), "Thread-6");
calThread1.start();
calThread2.start();
calThread3.start();
calThread4.start();
calThread5.start();
calThread6.start();
// 打印最终结果
storage.printTotal();
}
}
/**
* 计算器对象，负责计算start -> end
*/
class Calculate implements Runnable {
private Storage storage;
private long start;
public Calculate(long start, Storage storage) {
this.start = start;
this.storage = storage;
}
@Override
public void run() {
long num = start;
long sum = 0;
while (num <= 10000000) {
System.out.println(Thread.currentThread().getName() + " add num " + num);
sum += num;
num += 6;
}
// 线程计算完毕， 调用累加器进行累加
storage.push(sum);
}
}
/**
* 仓库对象，负责累加
*/
class Storage {
private long total = 0;
private int count = 0;
public synchronized void push(long sum) {
total += sum;
count++;
notifyAll();
}
public synchronized void printTotal() {
while (count < 6) {
try {
System.out.println(Thread.currentThread().getName() + " is wait");
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println("storage result = " + total);
}
}
思路二
线程异步返回：
利用线程池并发处理多个任务
使用Future+Callable获取异步执行结果
待线程池中所有任务结束，计算累加结果
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
/**
* 线程池计算累加数
*/
public class Accumulate {
public static void main(String[] args) {
// 建立线程池 与 动态结果数组
ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(6);
List<Future<Long>> resultList = new ArrayList<>();
// 定义10个任务分别负责一定范围内的元素累计
for (int i = 0; i < 10; i++) {
CalTask calTask = new CalTask(i*100000000+1, (i+1)*100000000);
Future<Long> result = executor.submit(calTask);
resultList.add(result);
}
// 每隔50毫秒遍历一遍所有动态结果，直到所有任务执行完毕
do {
System.out.printf("Main: 已经完成多少个任务: %d\n",executor.getCompletedTaskCount());
for (int i = 0; i < resultList.size(); i++) {
Future<Long> result = resultList.get(i);
System.out.printf("Main: Task %d is %s\n",i,result.isDone());
}
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
} while (executor.getCompletedTaskCount() < resultList.size());
// 若所有任务执行完毕，则对执行结果进行累计
long total = 0;
for (int i = 0; i < resultList.size(); i++) {
Future<Long> result = resultList.get(i);
long sum = 0;
try {
sum = result.get();
total += sum;
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
System.out.printf("total is: %d\n", total);
executor.shutdown();
}
}
class CalTask implements Callable<Long> {
private int startNum;
private int endNum;
public CalTask(int startNum, int endNum) {
this.startNum = startNum;
this.endNum = endNum;
}
@Override
public Long call() throws Exception {
long sum = 0;
for (int i = startNum; i <= endNum; i++) {
sum += i;
}
Thread.sleep(new Random().nextInt(100));
System.out.printf("%s: %d\n", Thread.currentThread().getName(), sum);
return sum;
}
}
网友评论