`

Java并发编程-Executor框架+实例

 
阅读更多


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. 


随机产生数和读取随机数文件的类如下: 

Java代码  收藏代码
  1. package my.concurrent.demo;  
  2.   
  3. import java.io.BufferedReader;  
  4. import java.io.BufferedWriter;  
  5. import java.io.File;  
  6. import java.io.FileNotFoundException;  
  7. import java.io.FileReader;  
  8. import java.io.FileWriter;  
  9. import java.io.IOException;  
  10. import java.util.Random;  
  11.   
  12. public class RandomUtil {  
  13.   
  14.     private static final int RANDOM_SEED= 10000;  
  15.   
  16.     private static final int SIZE = 10000;  
  17.   
  18.     /** 
  19.      * 产生10000万个随机数(范围1~9999),并将这些数据添加到指定文件中去。 
  20.      *  
  21.      * 例如: 
  22.      *  
  23.      * 1=7016 
  24.      * 2=7414 
  25.      * 3=3117 
  26.      * 4=6711 
  27.      * 5=5569 
  28.      * ... ...  
  29.      * 9993=1503 
  30.      * 9994=9528 
  31.      * 9995=9498 
  32.      * 9996=9123 
  33.      * 9997=6632 
  34.      * 9998=8801 
  35.      * 9999=9705 
  36.      * 10000=2900  
  37.      */  
  38.     public static void generatedRandomNbrs(String filepath) {  
  39.         Random random = new Random();  
  40.         BufferedWriter bw = null;  
  41.         try {  
  42.             bw = new BufferedWriter(new FileWriter(new File(filepath)));  
  43.             for (int i = 0; i < SIZE; i++) {  
  44.                 bw.write((i + 1) + "=" + random.nextInt(RANDOM_SEED));  
  45.                 bw.newLine();  
  46.             }  
  47.         } catch (IOException e) {  
  48.             // TODO Auto-generated catch block  
  49.             e.printStackTrace();  
  50.         } finally {  
  51.             if (null != bw) {  
  52.                 try {  
  53.                     bw.close();  
  54.                 } catch (IOException e) {  
  55.                     // TODO Auto-generated catch block  
  56.                     e.printStackTrace();  
  57.                 } finally {  
  58.                     bw = null;  
  59.                 }  
  60.             }  
  61.         }  
  62.     }  
  63.   
  64.     /** 
  65.      * 从指定文件中提取已经产生的随机数集 
  66.      */  
  67.     public static int[] populateValuesFromFile(String filepath) {  
  68.         BufferedReader br = null;  
  69.         int[] values = new int[SIZE];  
  70.   
  71.         try {  
  72.             br = new BufferedReader(new FileReader(new File(filepath)));  
  73.             int count = 0;  
  74.             String line = null;  
  75.             while (null != (line = br.readLine())) {  
  76.                 values[count++] = Integer.parseInt(line.substring(line  
  77.                         .indexOf("=") + 1));  
  78.             }  
  79.         } catch (FileNotFoundException e) {  
  80.             // TODO Auto-generated catch block  
  81.             e.printStackTrace();  
  82.         } catch (NumberFormatException e) {  
  83.             // TODO Auto-generated catch block  
  84.             e.printStackTrace();  
  85.         } catch (IOException e) {  
  86.             // TODO Auto-generated catch block  
  87.             e.printStackTrace();  
  88.         } finally {  
  89.             if (null != br) {  
  90.                 try {  
  91.                     br.close();  
  92.                 } catch (IOException e) {  
  93.                     // TODO Auto-generated catch block  
  94.                     e.printStackTrace();  
  95.                 } finally {  
  96.                     br = null;  
  97.                 }  
  98.             }  
  99.         }  
  100.         return values;  
  101.     }  
  102.   
  103. }  




编写一个Calculator 类, 实现Callable接口,计算指定数据集范围内的top 100. 

Java代码  收藏代码
  1. package my.concurrent.demo;  
  2.   
  3. import java.util.Arrays;  
  4. import java.util.concurrent.Callable;  
  5.   
  6. public class Calculator implements Callable<Integer[]> {  
  7.   
  8.     /** 待处理的数据 */  
  9.     private int[] values;  
  10.   
  11.     /** 起始索引 */  
  12.     private int startIndex;  
  13.   
  14.     /** 结束索引 */  
  15.     private int endIndex;  
  16.   
  17.     /** 
  18.      * @param values 
  19.      * @param startIndex 
  20.      * @param endIndex 
  21.      */  
  22.     public Calculator(int[] values, int startIndex, int endIndex) {  
  23.         this.values = values;  
  24.         this.startIndex = startIndex;  
  25.         this.endIndex = endIndex;  
  26.     }  
  27.   
  28.     public Integer[] call() throws Exception {  
  29.   
  30.         // 将指定范围的数据复制到指定的数组中去  
  31.         int[] subValues = new int[endIndex - startIndex + 1];  
  32.         System.arraycopy(values, startIndex, subValues, 0, endIndex  
  33.                 - startIndex + 1);  
  34.   
  35.         Arrays.sort(subValues);  
  36.   
  37.         // 将排序后的是数组数据,取出top 100 并返回。  
  38.         Integer[] top100 = new Integer[100];  
  39.         for (int i = 0; i < 100; i++) {  
  40.             top100[i] = subValues[subValues.length - i - 1];  
  41.         }  
  42.         return top100;  
  43.     }  
  44.   
  45.     /** 
  46.      * @return the values 
  47.      */  
  48.     public int[] getValues() {  
  49.         return values;  
  50.     }  
  51.   
  52.     /** 
  53.      * @param values 
  54.      *            the values to set 
  55.      */  
  56.     public void setValues(int[] values) {  
  57.         this.values = values;  
  58.     }  
  59.   
  60.     /** 
  61.      * @return the startIndex 
  62.      */  
  63.     public int getStartIndex() {  
  64.         return startIndex;  
  65.     }  
  66.   
  67.     /** 
  68.      * @param startIndex 
  69.      *            the startIndex to set 
  70.      */  
  71.     public void setStartIndex(int startIndex) {  
  72.         this.startIndex = startIndex;  
  73.     }  
  74.   
  75.     /** 
  76.      * @return the endIndex 
  77.      */  
  78.     public int getEndIndex() {  
  79.         return endIndex;  
  80.     }  
  81.   
  82.     /** 
  83.      * @param endIndex 
  84.      *            the endIndex to set 
  85.      */  
  86.     public void setEndIndex(int endIndex) {  
  87.         this.endIndex = endIndex;  
  88.     }  
  89.   
  90. }  



使用CompletionService实现 

Java代码  收藏代码
  1. package my.concurrent.demo;  
  2.   
  3. import java.util.Arrays;  
  4. import java.util.concurrent.ExecutionException;  
  5. import java.util.concurrent.ExecutorCompletionService;  
  6. import java.util.concurrent.ExecutorService;  
  7. import java.util.concurrent.Executors;  
  8.   
  9. public class ConcurrentCalculator {  
  10.   
  11.     private ExecutorService exec;  
  12.   
  13.     private ExecutorCompletionService<Integer[]> completionService;  
  14.   
  15.     private int availableProcessors = 0;  
  16.   
  17.     public ConcurrentCalculator() {  
  18.   
  19.         /* 
  20.          * 获取可用的处理器数量,并根据这个数量指定线程池的大小。 
  21.          */  
  22.         availableProcessors = populateAvailableProcessors();  
  23.         exec = Executors.newFixedThreadPool(availableProcessors);  
  24.   
  25.         completionService = new ExecutorCompletionService<Integer[]>(exec);  
  26.     }  
  27.   
  28.     /** 
  29.      * 获取10000个随机数中top 100的数。 
  30.      */  
  31.     public Integer[] top100(int[] values) {  
  32.   
  33.         /* 
  34.          * 用十个线程,每个线程处理1000个。 
  35.          */  
  36.         for (int i = 0; i < 10; i++) {  
  37.             completionService.submit(new Calculator(values, i * 1000,  
  38.                     i * 1000 + 1000 - 1));  
  39.         }  
  40.   
  41.         shutdown();  
  42.   
  43.         return populateTop100();  
  44.     }  
  45.   
  46.     /** 
  47.      * 计算top 100的数。 
  48.      *  
  49.      * 计算方法如下: 1. 初始化一个top 100的数组,数值都为0,作为当前的top 100. 2. 将这个当前的top 
  50.      * 100数组依次与每个线程产生的top 100数组比较,调整当前top 100的值。 
  51.      *  
  52.      */  
  53.     private Integer[] populateTop100() {  
  54.         Integer[] top100 = new Integer[100];  
  55.         for (int i = 0; i < 100; i++) {  
  56.             top100[i] = new Integer(0);  
  57.         }  
  58.   
  59.         for (int i = 0; i < 10; i++) {  
  60.             try {  
  61.                 adjustTop100(top100, completionService.take().get());  
  62.             } catch (InterruptedException e) {  
  63.                 e.printStackTrace();  
  64.             } catch (ExecutionException e) {  
  65.                 e.printStackTrace();  
  66.             }  
  67.         }  
  68.         return top100;  
  69.     }  
  70.   
  71.     /** 
  72.      * 将当前top 100数组和一个线程返回的top 100数组比较,并调整当前top 100数组的数据。 
  73.      */  
  74.     private void adjustTop100(Integer[] currentTop100, Integer[] subTop100) {  
  75.         Integer[] currentTop200 = new Integer[200];  
  76.   
  77.         System.arraycopy(currentTop100, 0, currentTop200, 0100);  
  78.         System.arraycopy(subTop100, 0, currentTop200, 100100);  
  79.   
  80.         Arrays.sort(currentTop200);  
  81.   
  82.         for (int i = 0; i < currentTop100.length; i++) {  
  83.             currentTop100[i] = currentTop200[currentTop200.length - i - 1];  
  84.         }  
  85.     }  
  86.   
  87.     /** 
  88.      * 关闭 executor 
  89.      */  
  90.     public void shutdown() {  
  91.         exec.shutdown();  
  92.     }  
  93.   
  94.     /** 
  95.      * 返回可以用的处理器个数 
  96.      */  
  97.     private int populateAvailableProcessors() {  
  98.         return Runtime.getRuntime().availableProcessors();  
  99.     }  
  100. }  



使用Callable,Future计算结果 

Java代码  收藏代码
  1. package my.concurrent.demo;  
  2.   
  3. import java.util.ArrayList;  
  4. import java.util.Arrays;  
  5. import java.util.List;  
  6. import java.util.concurrent.ExecutionException;  
  7. import java.util.concurrent.ExecutorService;  
  8. import java.util.concurrent.Executors;  
  9. import java.util.concurrent.Future;  
  10. import java.util.concurrent.FutureTask;  
  11.   
  12. public class ConcurrentCalculator2 {  
  13.   
  14.     private List<Future<Integer[]>> tasks = new ArrayList<Future<Integer[]>>();  
  15.   
  16.     private ExecutorService exec;  
  17.   
  18.     private int availableProcessors = 0;  
  19.   
  20.     public ConcurrentCalculator2() {  
  21.   
  22.         /* 
  23.          * 获取可用的处理器数量,并根据这个数量指定线程池的大小。 
  24.          */  
  25.         availableProcessors = populateAvailableProcessors();  
  26.         exec = Executors.newFixedThreadPool(availableProcessors);  
  27.   
  28.     }  
  29.   
  30.     /** 
  31.      * 获取10000个随机数中top 100的数。 
  32.      */  
  33.     public Integer[] top100(int[] values) {  
  34.   
  35.         /* 
  36.          * 用十个线程,每个线程处理1000个。 
  37.          */  
  38.         for (int i = 0; i < 10; i++) {  
  39.             FutureTask<Integer[]> task = new FutureTask<Integer[]>(  
  40.                     new Calculator(values, i * 1000, i * 1000 + 1000 - 1));  
  41.             tasks.add(task);  
  42.             if (!exec.isShutdown()) {  
  43.                 exec.submit(task);  
  44.             }  
  45.         }  
  46.   
  47.         shutdown();  
  48.   
  49.         return populateTop100();  
  50.     }  
  51.   
  52.     /** 
  53.      * 计算top 100的数。 
  54.      *  
  55.      * 计算方法如下: 1. 初始化一个top 100的数组,数值都为0,作为当前的top 100. 2. 将这个当前的top 
  56.      * 100数组依次与每个Task产生的top 100数组比较,调整当前top 100的值。 
  57.      *  
  58.      */  
  59.     private Integer[] populateTop100() {  
  60.         Integer[] top100 = new Integer[100];  
  61.         for (int i = 0; i < 100; i++) {  
  62.             top100[i] = new Integer(0);  
  63.         }  
  64.   
  65.         for (Future<Integer[]> task : tasks) {  
  66.             try {  
  67.                 adjustTop100(top100, task.get());  
  68.             } catch (InterruptedException e) {  
  69.                 // TODO Auto-generated catch block  
  70.                 e.printStackTrace();  
  71.             } catch (ExecutionException e) {  
  72.                 // TODO Auto-generated catch block  
  73.                 e.printStackTrace();  
  74.             }  
  75.         }  
  76.         return top100;  
  77.     }  
  78.   
  79.     /** 
  80.      * 将当前top 100数组和一个线程返回的top 100数组比较,并调整当前top 100数组的数据。 
  81.      */  
  82.     private void adjustTop100(Integer[] currentTop100, Integer[] subTop100) {  
  83.         Integer[] currentTop200 = new Integer[200];  
  84.         System.arraycopy(currentTop100, 0, currentTop200, 0100);  
  85.   
  86.         System.arraycopy(subTop100, 0, currentTop200, 100100);  
  87.   
  88.         Arrays.sort(currentTop200);  
  89.   
  90.         for (int i = 0; i < currentTop100.length; i++) {  
  91.             currentTop100[i] = currentTop200[currentTop200.length - i - 1];  
  92.         }  
  93.   
  94.     }  
  95.   
  96.     /** 
  97.      * 关闭executor 
  98.      */  
  99.     public void shutdown() {  
  100.         exec.shutdown();  
  101.     }  
  102.   
  103.     /** 
  104.      * 返回可以用的处理器个数 
  105.      */  
  106.     private int populateAvailableProcessors() {  
  107.         return Runtime.getRuntime().availableProcessors();  
  108.     }  
  109. }  



测试包括了三部分: 
1. 没有用Executor框架,用Arrays.sort直接计算,并从后往前取100个数。 
2. 使用CompletionService计算结果 
3. 使用Callable和Future计算结果 

测试代码如下: 

Java代码  收藏代码
  1. package my.concurrent.demo;  
  2.   
  3. import java.util.Arrays;  
  4.   
  5. public class Test {  
  6.   
  7.     private static final String FILE_PATH = "D:\\RandomNumber.txt";  
  8.   
  9.     public static void main(String[] args) {  
  10.         test();  
  11.     }  
  12.   
  13.     private static void test() {  
  14.         /* 
  15.          * 如果随机数已经存在文件中,可以不再调用此方法,除非想用新的随机数据。 
  16.          */  
  17.         //generateRandomNbrs();  
  18.           
  19.         process1();  
  20.   
  21.         process2();  
  22.   
  23.         process3();  
  24.   
  25.     }  
  26.   
  27.     private static void generateRandomNbrs() {  
  28.         RandomUtil.generatedRandomNbrs(FILE_PATH);  
  29.     }  
  30.   
  31.     private static void process1() {  
  32.         long start = System.currentTimeMillis();  
  33.         System.out.println("没有使用Executor框架,直接使用Arrays.sort获取top 100");  
  34.         printTop100(populateTop100(RandomUtil.populateValuesFromFile(FILE_PATH)));  
  35.         long end = System.currentTimeMillis();  
  36.         System.out.println((end - start) / 1000.0);  
  37.     }  
  38.   
  39.     private static void process2() {  
  40.         long start = System.currentTimeMillis();  
  41.   
  42.         System.out.println("使用ExecutorCompletionService获取top 100");  
  43.   
  44.         ConcurrentCalculator calculator = new ConcurrentCalculator();  
  45.         Integer[] top100 = calculator.top100(RandomUtil  
  46.                 .populateValuesFromFile(FILE_PATH));  
  47.         for (int i = 0; i < top100.length; i++) {  
  48.             System.out.println(String.format("top%d = %d", (i + 1), top100[i]));  
  49.         }  
  50.         long end = System.currentTimeMillis();  
  51.         System.out.println((end - start) / 1000.0);  
  52.     }  
  53.   
  54.     private static void process3() {  
  55.         long start = System.currentTimeMillis();  
  56.         System.out.println("使用FutureTask 获取top 100");  
  57.   
  58.         ConcurrentCalculator2 calculator2 = new ConcurrentCalculator2();  
  59.         Integer[] top100 = calculator2.top100(RandomUtil  
  60.                 .populateValuesFromFile(FILE_PATH));  
  61.         for (int i = 0; i < top100.length; i++) {  
  62.             System.out.println(String.format("top%d = %d", (i + 1), top100[i]));  
  63.         }  
  64.         long end = System.currentTimeMillis();  
  65.         System.out.println((end - start) / 1000.0);  
  66.     }  
  67.   
  68.     private static int[] populateTop100(int[] values) {  
  69.         Arrays.sort(values);  
  70.         int[] top100 = new int[100];  
  71.         int length = values.length;  
  72.         for (int i = 0; i < 100; i++) {  
  73.             top100[i] = values[length - 1 - i];  
  74.         }  
  75.         return top100;  
  76.     }  
  77.   
  78.     private static void printTop100(int[] top100) {  
  79.         for (int i = 0; i < top100.length; i++) {  
  80.             System.out.println(String.format("top%d = %d", (i + 1), top100[i]));  
  81.         }  
  82.     }  
  83.   
  84. }  



测试结果如下: 


分享到:
评论

相关推荐

    Java 并发编程实战.pdf

    《Java并发编程实战》这本书是关于Java语言中并发编程技术的经典著作。它详细介绍了如何在Java环境中有效地实现多线程程序和并发控制机制。在Java平台上,由于其本身提供了强大的并发编程支持,因此,掌握并发编程...

    Java并发编程利器:Executor框架深度解析与应用实践

    在现代Java应用开发中,多线程并发编程已成为提升程序性能的关键技术之一。Java通过引入Executor框架,为并发任务的执行提供了一种高效、灵活的管理机制。本文将深入探讨Executor框架的设计哲学、核心组件,并结合...

    Java-Executor并发框架.docx

    Java并发框架中的Executor服务是Java 1.5引入的核心组件,位于`java.util.concurrent`包下,极大地简化了多线程编程。Executor接口虽然历史悠久,但其重要性不言而喻,很多开发者对其背后的原理并不十分了解。本文将...

    《Java并发编程的艺术》

    《Java并发编程的艺术》内容涵盖Java并发编程机制的底层实现原理、Java内存模型、Java并发编程基础、Java中的锁、并发容器和框架、原子类、并发工具类、线程池、Executor框架等主题,每个主题都做了深入的讲解,同时...

    java并发编程:Executor、Executors、ExecutorService.docx

    Java并发编程中的Executor、Executors和ExecutorService是Java并发编程框架的重要组成部分,它们为开发者提供了高效管理和控制线程执行的工具。以下是对这些概念的详细解释: 1. Executor: Executor是一个接口,它...

    Java并发编程的艺术

    《Java并发编程的艺术》内容涵盖Java并发编程机制的底层实现原理、Java内存模型、Java并发编程基础、Java中的锁、并发容器和框架、原子类、并发工具类、线程池、Executor框架等主题,每个主题都做了深入的讲解,同时...

    Java并发编程实战

    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并发编程的艺术 作者:方腾飞 魏鹏 程晓明 著 丛书名:Java核心技术系列 出版日期 :2015-07-25 ISBN:978-7-111-50824-3 第1章介绍Java并发编程的挑战,向读者说明进入并发编程的世界可能会遇到哪些问题,以及如何...

    Java 并发编程实战(高清带目录).zip

    这本书高清且带有详细的目录,方便读者快速定位到所需的主题,是Java并发编程领域的重要参考资料。 并发编程是现代软件开发中的核心技能,尤其是在Java平台上,因为Java以其强大的并发支持而著称。本书涵盖了以下几...

    java并发编程实战.zip

    总的来说,《Java并发编程实战》这本书不仅提供了丰富的理论知识,还通过大量的实例源码帮助读者将理论付诸实践,是Java开发者提升并发编程技能的必备参考。通过学习书中的内容,开发者能够更好地理解和掌握Java并发...

    Java并发编程实践(Java Concurrency in Practice) (中英版)

    《Java并发编程实践》是Java开发者深入理解和应用并发编程的权威指南,这本书全面覆盖了Java并发编程的各种核心概念和技术,旨在帮助程序员编写出高效、安全的并发代码。书中的内容既包括理论知识,也包含丰富的实战...

    Java Executor 框架的实例详解

    Java Executor 框架是 Java 中一个很重要的并发编程模型,主要用于并发任务的执行和管理。Executor 框架的实例详解可以帮助开发者更好地理解和使用 Executor 框架,以下是 Java Executor 框架的实例详解的相关知识点...

    Executor框架使用详解

    Executor框架是Java并发编程的核心组件,它在Java 5中被引入,极大地简化了多线程编程。这个框架是基于`java.util.concurrent`包中的接口和类构建的,旨在提供线程池服务、任务调度以及并发执行任务的能力。Executor...

    JAVA并发编程实践 EN(全)

    1. **并发基础**:介绍Java并发编程的基本概念,包括线程、进程、同步与互斥、死锁等。读者将了解Java中Thread类和Runnable接口的使用,以及如何创建和管理线程。 2. **线程安全**:讨论线程不安全的代码示例,以及...

    java并发编程实战pdf及源码.zip

    《Java并发编程实战》这本书是Java并发编程领域的一本经典之作,它深入浅出地讲解了如何在Java环境中高效、安全地进行多线程编程。书中的内容涵盖了从基础概念到高级技术,包括线程的基本操作、同步机制、并发工具类...

    java并发编程实战源码-jcip-examples:java并发编程实战

    《Java并发编程实战》是Java并发编程领域的一本经典之作,由Brian Goetz等作者撰写。这本书深入浅出地介绍了Java并发编程的各种概念和技术,帮助开发者理解和掌握如何在多线程环境中编写高效、安全的代码。jcip-...

Global site tag (gtag.js) - Google Analytics