package com.hitachi.test.Thread;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;
import java.util.concurrent.*;

public class ConcurrentCalculator {
    private ExecutorService exec;
    private int cpuCoreNumber;
    private List> tasks = new ArrayList>();

// 内部类
    class SumCalculator implements Callable {
        private int[] numbers;
        private int start;
        private int end;

        public SumCalculator(final int[] numbers, int start, int end) {
            this.numbers = numbers;
            this.start = start;
            this.end = end;
        }

        public Long call() throws Exception {
            Long sum = 0l;
            for (int i = start; i < end; i++) {
                sum += numbers[i];
            }
            return sum;
        }
    }

    public ConcurrentCalculator() {
        cpuCoreNumber = Runtime.getRuntime().availableProcessors();
        exec = Executors.newFixedThreadPool(cpuCoreNumber);
    }

    public Long sum(final int[] numbers) {
        // 根据CPU核心个数拆分任务，创建FutureTask并提交到Executor
        System.out.println(cpuCoreNumber);
        for (int i = 0; i < cpuCoreNumber; i++) {
            int increment = numbers.length / cpuCoreNumber + 1;
            int start = increment * i;
            int end = increment * i + increment;
            if (end > numbers.length)
                end = numbers.length;
            SumCalculator subCalc = new SumCalculator(numbers, start, end);
            FutureTask task = new FutureTask(subCalc);
            tasks.add(task);
            if (!exec.isShutdown()) {
                exec.submit(task);
            }
        }
        return getResult();
    }

    /**
     * 迭代每个只任务，获得部分和，相加返回
     *
     * @return
     */
    public Long getResult() {
        Long result = 0l;
        for (Future task : tasks) {
            try {
                // 如果计算未完成则阻塞
                Long subSum = task.get();
                result += subSum;
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
        return result;
    }

    public void close() {
        exec.shutdown();
    }

    public static void main(String[] args){
        int[] numbers = new int[] { 1, 2, 3, 4, 5, 6, 7, 8, 10, 11, 100002,19999,1892384,1111114455,222222322,333333333,44444444 };
        ConcurrentCalculator calc = new ConcurrentCalculator();
        System.out.println(Calendar.getInstance().getTimeInMillis());
        Long sum = calc.sum(numbers);
        System.out.println(Calendar.getInstance().getTimeInMillis());
        System.out.println(sum);
        calc.close();
    }

}

An object that executes submitted Runnable tasks. This interface provides a way of decoupling task submission from the mechanics of how each task will be run, including details of thread use, scheduling, etc. An Executor is normally used instead of explicitly creating threads. For example, rather than invoking new Thread(new(RunnableTask())).start() for each of a set of tasks, you might use: Executor executor = anExecutor; executor.execute(new RunnableTask1()); executor.execute(new RunnableTask2());

ITPub博客

多线程Executor

最新文章