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. 


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

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]));
        }
    }

}

测试结果如下：