Executor框架是指java 5中引入的一系列并发库中与executor相关的一些功能类,其中包括线程池,Executor,Executors,ExecutorService,CompletionService,Future,Callable等。运用该框架能够很好的将任务分成一个个的子任务,使并发编程变得方便。该框架的类图(方法并没有都表示出来)如下:
创建线程池的介绍,摘自http://mshijie.iteye.com/blog/366591
创建线程池
Executors类,提供了一系列工厂方法用于创先线程池,返回的线程池都实现了ExecutorService接口。
public static ExecutorService newFixedThreadPool(int nThreads)
创建固定数目线程的线程池。
public static ExecutorService newCachedThreadPool()
创建一个可缓存的线程池,调用execute 将重用以前构造的线程(如果线程可用)。如果现有线程没有可用的,则创建一个新线程并添加到池中。终止并从缓存中移除那些已有 60 秒钟未被使用的线程。
public static ExecutorService newSingleThreadExecutor()
创建一个单线程化的Executor。
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)
创建一个支持定时及周期性的任务执行的线程池,多数情况下可用来替代Timer类。
本文主要是用该Executor框架来完成一个任务:求出10000个随机数据中的top 100.
Note:本文只是用Executor来做一个例子,并不是用最好的办法去求10000个数中最大的100个数。
具体的实现如下:
1. 随机产生10000个数(范围1~9999),并存放在一个文件中。
2. 读取该文件的数值,并存放在一个数组中。
3. 采用Executor框架,进行并发操作,将10000个数据用10个线程来做,每个线程完成1000=(10000/10)个数据的top 100操作。
4. 将10个线程返回的各个top 100数据,重新计算,得出最后的10000个数据的top 100.
随机产生数和读取随机数文件的类如下:
- package my.concurrent.demo;
- import java.io.BufferedReader;
- import java.io.BufferedWriter;
- import java.io.File;
- import java.io.FileNotFoundException;
- import java.io.FileReader;
- import java.io.FileWriter;
- import java.io.IOException;
- import java.util.Random;
- public class RandomUtil {
- private static final int RANDOM_SEED= 10000;
- private static final int SIZE = 10000;
- /**
- * 产生10000万个随机数(范围1~9999),并将这些数据添加到指定文件中去。
- *
- * 例如:
- *
- * 1=7016
- * 2=7414
- * 3=3117
- * 4=6711
- * 5=5569
- * ... ...
- * 9993=1503
- * 9994=9528
- * 9995=9498
- * 9996=9123
- * 9997=6632
- * 9998=8801
- * 9999=9705
- * 10000=2900
- */
- public static void generatedRandomNbrs(String filepath) {
- Random random = new Random();
- BufferedWriter bw = null;
- try {
- bw = new BufferedWriter(new FileWriter(new File(filepath)));
- for (int i = 0; i < SIZE; i++) {
- bw.write((i + 1) + "=" + random.nextInt(RANDOM_SEED));
- bw.newLine();
- }
- } catch (IOException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- } finally {
- if (null != bw) {
- try {
- bw.close();
- } catch (IOException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- } finally {
- bw = null;
- }
- }
- }
- }
- /**
- * 从指定文件中提取已经产生的随机数集
- */
- public static int[] populateValuesFromFile(String filepath) {
- BufferedReader br = null;
- int[] values = new int[SIZE];
- try {
- br = new BufferedReader(new FileReader(new File(filepath)));
- int count = 0;
- String line = null;
- while (null != (line = br.readLine())) {
- values[count++] = Integer.parseInt(line.substring(line
- .indexOf("=") + 1));
- }
- } catch (FileNotFoundException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- } catch (NumberFormatException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- } catch (IOException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- } finally {
- if (null != br) {
- try {
- br.close();
- } catch (IOException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- } finally {
- br = null;
- }
- }
- }
- return values;
- }
- }
编写一个Calculator 类, 实现Callable接口,计算指定数据集范围内的top 100.
- package my.concurrent.demo;
- import java.util.Arrays;
- import java.util.concurrent.Callable;
- public class Calculator implements Callable<Integer[]> {
- /** 待处理的数据 */
- private int[] values;
- /** 起始索引 */
- private int startIndex;
- /** 结束索引 */
- private int endIndex;
- /**
- * @param values
- * @param startIndex
- * @param endIndex
- */
- public Calculator(int[] values, int startIndex, int endIndex) {
- this.values = values;
- this.startIndex = startIndex;
- this.endIndex = endIndex;
- }
- public Integer[] call() throws Exception {
- // 将指定范围的数据复制到指定的数组中去
- int[] subValues = new int[endIndex - startIndex + 1];
- System.arraycopy(values, startIndex, subValues, 0, endIndex
- - startIndex + 1);
- Arrays.sort(subValues);
- // 将排序后的是数组数据,取出top 100 并返回。
- Integer[] top100 = new Integer[100];
- for (int i = 0; i < 100; i++) {
- top100[i] = subValues[subValues.length - i - 1];
- }
- return top100;
- }
- /**
- * @return the values
- */
- public int[] getValues() {
- return values;
- }
- /**
- * @param values
- * the values to set
- */
- public void setValues(int[] values) {
- this.values = values;
- }
- /**
- * @return the startIndex
- */
- public int getStartIndex() {
- return startIndex;
- }
- /**
- * @param startIndex
- * the startIndex to set
- */
- public void setStartIndex(int startIndex) {
- this.startIndex = startIndex;
- }
- /**
- * @return the endIndex
- */
- public int getEndIndex() {
- return endIndex;
- }
- /**
- * @param endIndex
- * the endIndex to set
- */
- public void setEndIndex(int endIndex) {
- this.endIndex = endIndex;
- }
- }
使用CompletionService实现
- package my.concurrent.demo;
- import java.util.Arrays;
- import java.util.concurrent.ExecutionException;
- import java.util.concurrent.ExecutorCompletionService;
- import java.util.concurrent.ExecutorService;
- import java.util.concurrent.Executors;
- public class ConcurrentCalculator {
- private ExecutorService exec;
- private ExecutorCompletionService<Integer[]> completionService;
- private int availableProcessors = 0;
- public ConcurrentCalculator() {
- /*
- * 获取可用的处理器数量,并根据这个数量指定线程池的大小。
- */
- availableProcessors = populateAvailableProcessors();
- exec = Executors.newFixedThreadPool(availableProcessors);
- completionService = new ExecutorCompletionService<Integer[]>(exec);
- }
- /**
- * 获取10000个随机数中top 100的数。
- */
- public Integer[] top100(int[] values) {
- /*
- * 用十个线程,每个线程处理1000个。
- */
- for (int i = 0; i < 10; i++) {
- completionService.submit(new Calculator(values, i * 1000,
- i * 1000 + 1000 - 1));
- }
- shutdown();
- return populateTop100();
- }
- /**
- * 计算top 100的数。
- *
- * 计算方法如下: 1. 初始化一个top 100的数组,数值都为0,作为当前的top 100. 2. 将这个当前的top
- * 100数组依次与每个线程产生的top 100数组比较,调整当前top 100的值。
- *
- */
- private Integer[] populateTop100() {
- Integer[] top100 = new Integer[100];
- for (int i = 0; i < 100; i++) {
- top100[i] = new Integer(0);
- }
- for (int i = 0; i < 10; i++) {
- try {
- adjustTop100(top100, completionService.take().get());
- } catch (InterruptedException e) {
- e.printStackTrace();
- } catch (ExecutionException e) {
- e.printStackTrace();
- }
- }
- return top100;
- }
- /**
- * 将当前top 100数组和一个线程返回的top 100数组比较,并调整当前top 100数组的数据。
- */
- private void adjustTop100(Integer[] currentTop100, Integer[] subTop100) {
- Integer[] currentTop200 = new Integer[200];
- System.arraycopy(currentTop100, 0, currentTop200, 0, 100);
- System.arraycopy(subTop100, 0, currentTop200, 100, 100);
- Arrays.sort(currentTop200);
- for (int i = 0; i < currentTop100.length; i++) {
- currentTop100[i] = currentTop200[currentTop200.length - i - 1];
- }
- }
- /**
- * 关闭 executor
- */
- public void shutdown() {
- exec.shutdown();
- }
- /**
- * 返回可以用的处理器个数
- */
- private int populateAvailableProcessors() {
- return Runtime.getRuntime().availableProcessors();
- }
- }
使用Callable,Future计算结果
- package my.concurrent.demo;
- import java.util.ArrayList;
- import java.util.Arrays;
- import java.util.List;
- import java.util.concurrent.ExecutionException;
- import java.util.concurrent.ExecutorService;
- import java.util.concurrent.Executors;
- import java.util.concurrent.Future;
- import java.util.concurrent.FutureTask;
- public class ConcurrentCalculator2 {
- private List<Future<Integer[]>> tasks = new ArrayList<Future<Integer[]>>();
- private ExecutorService exec;
- private int availableProcessors = 0;
- public ConcurrentCalculator2() {
- /*
- * 获取可用的处理器数量,并根据这个数量指定线程池的大小。
- */
- availableProcessors = populateAvailableProcessors();
- exec = Executors.newFixedThreadPool(availableProcessors);
- }
- /**
- * 获取10000个随机数中top 100的数。
- */
- public Integer[] top100(int[] values) {
- /*
- * 用十个线程,每个线程处理1000个。
- */
- for (int i = 0; i < 10; i++) {
- FutureTask<Integer[]> task = new FutureTask<Integer[]>(
- new Calculator(values, i * 1000, i * 1000 + 1000 - 1));
- tasks.add(task);
- if (!exec.isShutdown()) {
- exec.submit(task);
- }
- }
- shutdown();
- return populateTop100();
- }
- /**
- * 计算top 100的数。
- *
- * 计算方法如下: 1. 初始化一个top 100的数组,数值都为0,作为当前的top 100. 2. 将这个当前的top
- * 100数组依次与每个Task产生的top 100数组比较,调整当前top 100的值。
- *
- */
- private Integer[] populateTop100() {
- Integer[] top100 = new Integer[100];
- for (int i = 0; i < 100; i++) {
- top100[i] = new Integer(0);
- }
- for (Future<Integer[]> task : tasks) {
- try {
- adjustTop100(top100, task.get());
- } catch (InterruptedException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- } catch (ExecutionException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
- }
- return top100;
- }
- /**
- * 将当前top 100数组和一个线程返回的top 100数组比较,并调整当前top 100数组的数据。
- */
- private void adjustTop100(Integer[] currentTop100, Integer[] subTop100) {
- Integer[] currentTop200 = new Integer[200];
- System.arraycopy(currentTop100, 0, currentTop200, 0, 100);
- System.arraycopy(subTop100, 0, currentTop200, 100, 100);
- Arrays.sort(currentTop200);
- for (int i = 0; i < currentTop100.length; i++) {
- currentTop100[i] = currentTop200[currentTop200.length - i - 1];
- }
- }
- /**
- * 关闭executor
- */
- public void shutdown() {
- exec.shutdown();
- }
- /**
- * 返回可以用的处理器个数
- */
- private int populateAvailableProcessors() {
- return Runtime.getRuntime().availableProcessors();
- }
- }
测试包括了三部分:
1. 没有用Executor框架,用Arrays.sort直接计算,并从后往前取100个数。
2. 使用CompletionService计算结果
3. 使用Callable和Future计算结果
测试代码如下:
- package my.concurrent.demo;
- import java.util.Arrays;
- public class Test {
- private static final String FILE_PATH = "D:\\RandomNumber.txt";
- public static void main(String[] args) {
- test();
- }
- private static void test() {
- /*
- * 如果随机数已经存在文件中,可以不再调用此方法,除非想用新的随机数据。
- */
- //generateRandomNbrs();
- process1();
- process2();
- process3();
- }
- private static void generateRandomNbrs() {
- RandomUtil.generatedRandomNbrs(FILE_PATH);
- }
- private static void process1() {
- long start = System.currentTimeMillis();
- System.out.println("没有使用Executor框架,直接使用Arrays.sort获取top 100");
- printTop100(populateTop100(RandomUtil.populateValuesFromFile(FILE_PATH)));
- long end = System.currentTimeMillis();
- System.out.println((end - start) / 1000.0);
- }
- private static void process2() {
- long start = System.currentTimeMillis();
- System.out.println("使用ExecutorCompletionService获取top 100");
- ConcurrentCalculator calculator = new ConcurrentCalculator();
- Integer[] top100 = calculator.top100(RandomUtil
- .populateValuesFromFile(FILE_PATH));
- for (int i = 0; i < top100.length; i++) {
- System.out.println(String.format("top%d = %d", (i + 1), top100[i]));
- }
- long end = System.currentTimeMillis();
- System.out.println((end - start) / 1000.0);
- }
- private static void process3() {
- long start = System.currentTimeMillis();
- System.out.println("使用FutureTask 获取top 100");
- ConcurrentCalculator2 calculator2 = new ConcurrentCalculator2();
- Integer[] top100 = calculator2.top100(RandomUtil
- .populateValuesFromFile(FILE_PATH));
- for (int i = 0; i < top100.length; i++) {
- System.out.println(String.format("top%d = %d", (i + 1), top100[i]));
- }
- long end = System.currentTimeMillis();
- System.out.println((end - start) / 1000.0);
- }
- private static int[] populateTop100(int[] values) {
- Arrays.sort(values);
- int[] top100 = new int[100];
- int length = values.length;
- for (int i = 0; i < 100; i++) {
- top100[i] = values[length - 1 - i];
- }
- return top100;
- }
- private static void printTop100(int[] top100) {
- for (int i = 0; i < top100.length; i++) {
- System.out.println(String.format("top%d = %d", (i + 1), top100[i]));
- }
- }
- }
测试结果如下:
相关推荐
《Java并发编程实战》这本书是关于Java语言中并发编程技术的经典著作。它详细介绍了如何在Java环境中有效地实现多线程程序和并发控制机制。在Java平台上,由于其本身提供了强大的并发编程支持,因此,掌握并发编程...
在现代Java应用开发中,多线程并发编程已成为提升程序性能的关键技术之一。Java通过引入Executor框架,为并发任务的执行提供了一种高效、灵活的管理机制。本文将深入探讨Executor框架的设计哲学、核心组件,并结合...
Java并发框架中的Executor服务是Java 1.5引入的核心组件,位于`java.util.concurrent`包下,极大地简化了多线程编程。Executor接口虽然历史悠久,但其重要性不言而喻,很多开发者对其背后的原理并不十分了解。本文将...
《Java并发编程的艺术》内容涵盖Java并发编程机制的底层实现原理、Java内存模型、Java并发编程基础、Java中的锁、并发容器和框架、原子类、并发工具类、线程池、Executor框架等主题,每个主题都做了深入的讲解,同时...
Java并发编程中的Executor、Executors和ExecutorService是Java并发编程框架的重要组成部分,它们为开发者提供了高效管理和控制线程执行的工具。以下是对这些概念的详细解释: 1. Executor: Executor是一个接口,它...
《Java并发编程的艺术》内容涵盖Java并发编程机制的底层实现原理、Java内存模型、Java并发编程基础、Java中的锁、并发容器和框架、原子类、并发工具类、线程池、Executor框架等主题,每个主题都做了深入的讲解,同时...
6.2 Executor框架 6.2.1 示例:基于Executor的Web服务器 6.2.2 执行策略 6.2.3 线程池 6.2.4 Executor的生命周期 6.2.5 延迟任务与周期任务 6.3 找出可利用的并行性 6.3.1 示例:串行的页面渲染器 6.3.2 ...
Java并发编程的艺术 作者:方腾飞 魏鹏 程晓明 著 丛书名:Java核心技术系列 出版日期 :2015-07-25 ISBN:978-7-111-50824-3 第1章介绍Java并发编程的挑战,向读者说明进入并发编程的世界可能会遇到哪些问题,以及如何...
这本书高清且带有详细的目录,方便读者快速定位到所需的主题,是Java并发编程领域的重要参考资料。 并发编程是现代软件开发中的核心技能,尤其是在Java平台上,因为Java以其强大的并发支持而著称。本书涵盖了以下几...
总的来说,《Java并发编程实战》这本书不仅提供了丰富的理论知识,还通过大量的实例源码帮助读者将理论付诸实践,是Java开发者提升并发编程技能的必备参考。通过学习书中的内容,开发者能够更好地理解和掌握Java并发...
《Java并发编程实践》是Java开发者深入理解和应用并发编程的权威指南,这本书全面覆盖了Java并发编程的各种核心概念和技术,旨在帮助程序员编写出高效、安全的并发代码。书中的内容既包括理论知识,也包含丰富的实战...
Java Executor 框架是 Java 中一个很重要的并发编程模型,主要用于并发任务的执行和管理。Executor 框架的实例详解可以帮助开发者更好地理解和使用 Executor 框架,以下是 Java Executor 框架的实例详解的相关知识点...
Executor框架是Java并发编程的核心组件,它在Java 5中被引入,极大地简化了多线程编程。这个框架是基于`java.util.concurrent`包中的接口和类构建的,旨在提供线程池服务、任务调度以及并发执行任务的能力。Executor...
1. **并发基础**:介绍Java并发编程的基本概念,包括线程、进程、同步与互斥、死锁等。读者将了解Java中Thread类和Runnable接口的使用,以及如何创建和管理线程。 2. **线程安全**:讨论线程不安全的代码示例,以及...
《Java并发编程实战》这本书是Java并发编程领域的一本经典之作,它深入浅出地讲解了如何在Java环境中高效、安全地进行多线程编程。书中的内容涵盖了从基础概念到高级技术,包括线程的基本操作、同步机制、并发工具类...
《Java并发编程实战》是Java并发编程领域的一本经典之作,由Brian Goetz等作者撰写。这本书深入浅出地介绍了Java并发编程的各种概念和技术,帮助开发者理解和掌握如何在多线程环境中编写高效、安全的代码。jcip-...