java多线程之线程池

public interface Executor {
 
//提交运行任务，参数为Runnable接口对象，无返回值
    void execute(Runnable command);
}

public interface ExecutorService extends Executor {
 
    //关闭线程池，线程池中不再接受新提交的任务，但是之前提交的任务继续运行，直到完成
    void shutdown();
 
    //关闭线程池，线程池中不再接受新提交的任务，会尝试停止线程池中正在执行的任务。
    List<Runnable> shutdownNow();
//判断线程池是否已经关闭
   
    boolean isShutdown();
//判断线程池中的所有任务是否结束，只有在调用shutdown或者shutdownNow方法之后调用此方法才会返回true。
   
    boolean isTerminated();
//等待线程池中的所有任务执行结束，并设置超时时间
   
    boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;
//提交一个Callable接口类型的任务，返回一个Future类型的结果
    
    <T> Future<T> submit(Callable<T> task);
 
    //提交一个Callable接口类型的任务，并且给定一个泛型类型的接收结果数据参数，返回一个Future类型的结果
    <T> Future<T> submit(Runnable task, T result);
//提交一个Runnable接口类型的任务，返回一个Future类型的结果
    
    Future<?> submit(Runnable task);
//批量提交任务并获得他们的future，Task列表与Future列表一一对应
    
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;
 
    //批量提交任务并获得他们的future，并限定处理所有任务的时间
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                  long timeout, TimeUnit unit)
        throws InterruptedException;
//批量提交任务并获得一个已经成功执行的任务的结果
    
    <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;
//批量提交任务并获得一个已经成功执行的任务的结果，并限定处理任务的时间
    
    <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                    long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

public abstract class AbstractExecutorService implements ExecutorService {
 
    
//RunnableFuture类用于获取执行结果，在实际使用时，我们经常使用的是它的子类FutureTask，newTaskFor方法的作用就是将任务封装成FutureTask对象，后续将FutureTask对象提交到线程池。
    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
        return new FutureTask<T>(runnable, value);
    }
   
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        return new FutureTask<T>(callable);
    }
 
   
    public Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<Void> ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
    }
 
    public <T> Future<T> submit(Runnable task, T result) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task, result);
        execute(ftask);
        return ftask;
    }
 
   
    public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }
 
   
    private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
                              boolean timed, long nanos)
        throws InterruptedException, ExecutionException, TimeoutException {
//提交的任务为空，抛出空指针异常
        if (tasks == null)
            throw new NullPointerException();
//记录待执行的任务的剩余数量
        int ntasks = tasks.size();
//任务集合中的数据为空，抛出非法参数异常
        if (ntasks == 0)
            throw new IllegalArgumentException();
        ArrayList<Future<T>> futures = new ArrayList<Future<T>>(ntasks);
//以当前实例对象作为参数构建ExecutorCompletionService对象
// ExecutorCompletionService负责执行任务，后面调用用poll返回第一个执行结果
        ExecutorCompletionService<T> ecs =
            new ExecutorCompletionService<T>(this);
 
       
 
        try {
            // 记录可能抛出的执行异常
            ExecutionException ee = null;
// 初始化超时时间
            final long deadline = timed ? System.nanoTime() + nanos : 0L;
            Iterator<? extends Callable<T>> it = tasks.iterator();
 
            //提交任务，并将返回的结果数据添加到futures集合中
//提交一个任务主要是确保在进入循环之前开始一个任务
            futures.add(ecs.submit(it.next()));
            --ntasks;
//记录正在执行的任务数量
            int active = 1;
 
            for (;;) {
//从完成任务的BlockingQueue队列中获取并移除下一个将要完成的任务的结果。
//如果BlockingQueue队列中中的数据为空，则返回null
//这里的poll()方法是非阻塞方法
                Future<T> f = ecs.poll();
//获取的结果为空
                if (f == null) {
//集合中仍有未执行的任务数量
                    if (ntasks > 0) {
//未执行的任务数量减1
 
                        --ntasks;
//提交完成并将结果添加到futures集合中
                        futures.add(ecs.submit(it.next()));
//正在执行的任务数量加•1
 
                        ++active;
                    }
//所有任务执行完成，并且返回了结果数据，则退出循环
//之所以处理active为0的情况，是因为poll()方法是非阻塞方法，可能导致未返回结果时active为0
                    else if (active == 0)
                        break;
//如果timed为true，则执行获取结果数据时设置超时时间，也就是超时获取结果表示
                    else if (timed) {
                        f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
                        if (f == null)
                            throw new TimeoutException();
                        nanos = deadline - System.nanoTime();
                    }
//没有设置超时，并且所有任务都被提交了，则一直阻塞，直到返回一个执行结果
                    else
                        f = ecs.take();
                }
//获取到执行结果，则将正在执行的任务减1，从Future中获取结果并返回
                if (f != null) {
                    --active;
                    try {
                        return f.get();
                    } catch (ExecutionException eex) {
                        ee = eex;
                    } catch (RuntimeException rex) {
                        ee = new ExecutionException(rex);
                    }
                }
            }
 
            if (ee == null)
                ee = new ExecutionException();
            throw ee;
 
        } finally {
//如果从所有执行的任务中获取到一个结果数据，则取消所有执行的任务，不再向下执行
            for (int i = 0, size = futures.size(); i < size; i++)
                futures.get(i).cancel(true);
        }
    }
 
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException {
        try {
            return doInvokeAny(tasks, false, 0);
        } catch (TimeoutException cannotHappen) {
            assert false;
            return null;
        }
    }
 
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                           long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
        return doInvokeAny(tasks, true, unit.toNanos(timeout));
    }
 
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException {
        if (tasks == null)
            throw new NullPointerException();
        ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
        boolean done = false;
        try {
            for (Callable<T> t : tasks) {
                RunnableFuture<T> f = newTaskFor(t);
                futures.add(f);
                execute(f);
            }
            for (int i = 0, size = futures.size(); i < size; i++) {
                Future<T> f = futures.get(i);
                if (!f.isDone()) {
                    try {
                        f.get();
                    } catch (CancellationException ignore) {
                    } catch (ExecutionException ignore) {
                    }
                }
            }
            done = true;
            return futures;
        } finally {
            if (!done)
                for (int i = 0, size = futures.size(); i < size; i++)
                    futures.get(i).cancel(true);
        }
    }
 
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                         long timeout, TimeUnit unit)
        throws InterruptedException {
        if (tasks == null)
            throw new NullPointerException();
        long nanos = unit.toNanos(timeout);
        ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
        boolean done = false;
        try {
            for (Callable<T> t : tasks)
                futures.add(newTaskFor(t));
 
            final long deadline = System.nanoTime() + nanos;
            final int size = futures.size();
 
            // Interleave time checks and calls to execute in case
            // executor doesn't have any/much parallelism.
            for (int i = 0; i < size; i++) {
                execute((Runnable)futures.get(i));
                nanos = deadline - System.nanoTime();
                if (nanos <= 0L)
                    return futures;
            }
 
            for (int i = 0; i < size; i++) {
                Future<T> f = futures.get(i);
                if (!f.isDone()) {
                    if (nanos <= 0L)
                        return futures;
                    try {
                        f.get(nanos, TimeUnit.NANOSECONDS);
                    } catch (CancellationException ignore) {
                    } catch (ExecutionException ignore) {
                    } catch (TimeoutException toe) {
                        return futures;
                    }
                    nanos = deadline - System.nanoTime();
                }
            }
            done = true;
            return futures;
        } finally {
            if (!done)
                for (int i = 0, size = futures.size(); i < size; i++)
                    futures.get(i).cancel(true);
        }
    }
 
}

private static ThreadFactory threadFactory = new ThreadFactoryBuilder().build();
 
private static ExecutorService executorService = new ThreadPoolExecutor(10, 10, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10), threadFactory, new ThreadPoolExecutor.AbortPolicy());

public class MyTest {
    public static void main(String[] args) throws InterruptedException {
        List<Integer> list = new Vector<>();
        ExecutorService executorService = Executors.newFixedThreadPool(1000);
        for (int i = 0; i < 1000; i++) {
            final  int index = i;
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName()+"-index="+index);
                   try {
                       Thread.sleep(1000);
                       list.add(index);
                   }catch (Exception e){
                       e.printStackTrace();
                   }
                }
            });
        }
        //不在接收新的任务
        executorService.shutdown();
        // 等待所有线程执行完毕
        executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
        System.out.println("all element size====================================================="+list.size());
    }
}

 public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
// 如果线程数小于基本线程数，则创建线程并执行当前任务
        if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
// 如线程数大于等于基本线程数或线程创建失败，则将当前任务放到工作队列中。
            if (runState == RUNNING && workQueue.offer(command)) {
                if (runState != RUNNING || poolSize == 0)
                    ensureQueuedTaskHandled(command);
            }
// 如果线程池不处于运行中或任务无法放入队列，并且当前线程数量小于最大允许的线程数量，
// 则创建一个线程执行任务。
            else if (!addIfUnderMaximumPoolSize(command))
// 抛出RejectedExecutionException异常
                reject(command); // is shutdown or saturated
        }
    }

public void singleThreadExecutorDemo(){
    ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
    for (int i = 0; i < 3; i++) {
        final int index = i;
 
        singleThreadExecutor.execute(new Runnable() {
 
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName()+", index="+index);
            }
        });
 
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
 
pool-1-thread-1, index=0
pool-1-thread-1, index=1
pool-1-thread-1, index=2

public void cachedThreadPoolDemo(){
    ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
    for (int i = 0; i < 5; i++) {
        final int index = i;
 
        cachedThreadPool.execute(new Runnable() {
 
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName()+", index="+index);
            }
        });
 
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
 
pool-1-thread-1, index=0
pool-1-thread-1, index=1
pool-1-thread-1, index=2
pool-1-thread-1, index=3
pool-1-thread-1, index=4
    

public void fixedThreadPoolDemo(){
    ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
    for (int i = 0; i < 6; i++) {
        final int index = i;
 
        fixedThreadPool.execute(new Runnable() {
 
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName()+", index="+index);
            }
        });
 
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
 
pool-1-thread-1, index=0
pool-1-thread-2, index=1
pool-1-thread-3, index=2
pool-1-thread-1, index=3
pool-1-thread-2, index=4
pool-1-thread-3, index=5

public void scheduledThreadPoolDemo(){
    ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(3);
    //定时执行一次的任务，延迟1s后执行
    scheduledThreadPool.schedule(new Runnable() {
 
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName()+", delay 1s");
        }
    }, 1, TimeUnit.SECONDS);
 
    //周期性地执行任务，延迟2s后，每3s一次地周期性执行任务
    scheduledThreadPool.scheduleAtFixedRate(new Runnable() {
 
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName()+", every 3s");
        }
    }, 2, 3, TimeUnit.SECONDS);
}
 
pool-1-thread-1, delay 1s
pool-1-thread-1, every 3s
pool-1-thread-2, every 3s
pool-1-thread-2, every 3s

// 本⽂所有代码都引⾃Java 8
public final ForkJoinTask<V> fork() {
 Thread t;
 // ForkJoinWorkerThread是执⾏ForkJoinTask的专有线程，由ForkJoinPool管理
 // 先判断当前线程是否是ForkJoin专有线程，如果是，则将任务push到当前线程所负责的队列⾥去
 if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
 ((ForkJoinWorkerThread)t).workQueue.push(this);
 else
 // 如果不是则将线程加⼊队列
 // 没有显式创建ForkJoinPool的时候⾛这⾥，提交任务到默认的common线程池中
 ForkJoinPool.common.externalPush(this);
 return this;
}

    public class ForkJoinPool extends AbstractExecutorService {
        // 任务队列
        volatile WorkQueue[] workQueues;
        // 线程的运⾏状态
        volatile int runState;
        // 创建ForkJoinWorkerThread的默认⼯⼚，可以通过构造函数重写
        public static final ForkJoinWorkerThreadFactory defaultForkJoinWorkerThread
        // 公⽤的线程池，其运⾏状态不受shutdown()和shutdownNow()的影响
        static final ForkJoinPool common;
        // 私有构造⽅法，没有任何安全检查和参数校验，由makeCommonPool直接调⽤
        // 其他构造⽅法都是源⾃于此⽅法
        // parallelism: 并⾏度，
        // 默认调⽤java.lang.Runtime.availableProcessors() ⽅法返回可⽤处理器的数量
        private ForkJoinPool(int parallelism,
                             ForkJoinWorkerThreadFactory factory, // ⼯作线程⼯⼚
                             UncaughtExceptionHandler handler, // 拒绝任务的handler
                             int mode, // 同步模式
                             String workerNamePrefix) { // 线程名prefix
            this.workerNamePrefix = workerNamePrefix;
            this.factory = factory;
            this.ueh = handler;
            this.config = (parallelism & SMASK) | mode;
            long np = (long)(-parallelism); // offset ctl counts
            this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK)
        }
    }

public class Main {
    public static void main(String[] args) throws Exception {
        // 创建2000个随机数组成的数组:
        long[] array = new long[2000];
        long expectedSum = 0;
        for (int i = 0; i < array.length; i++) {
            array[i] = random();
            expectedSum += array[i];
        }
        System.out.println("Expected sum: " + expectedSum);
        // fork/join:
        ForkJoinTask<Long> task = new SumTask(array, 0, array.length);
        long startTime = System.currentTimeMillis();
        Long result = ForkJoinPool.commonPool().invoke(task);
        long endTime = System.currentTimeMillis();
        System.out.println("Fork/join sum: " + result + " in " + (endTime - startTime) + " ms.");
    }
 
    static Random random = new Random(0);
 
    static long random() {
        return random.nextInt(10000);
    }
}
 
class SumTask extends RecursiveTask<Long> {
    static final int THRESHOLD = 500;
    long[] array;
    int start;
    int end;
 
    SumTask(long[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }
 
    @Override
    protected Long compute() {
        if (end - start <= THRESHOLD) {
            // 如果任务足够小,直接计算:
            long sum = 0;
            for (int i = start; i < end; i++) {
                sum += this.array[i];
                // 故意放慢计算速度:
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                }
            }
            return sum;
        }
        // 任务太大,一分为二:
        int middle = (end + start) / 2;
        System.out.println(String.format("split %d~%d ==> %d~%d, %d~%d", start, end, start, middle, middle, end));
        SumTask subtask1 = new SumTask(this.array, start, middle);
        SumTask subtask2 = new SumTask(this.array, middle, end);
        invokeAll(subtask1, subtask2);
        Long subresult1 = subtask1.join();
        Long subresult2 = subtask2.join();
        Long result = subresult1 + subresult2;
        System.out.println("result = " + subresult1 + " + " + subresult2 + " ==> " + result);
        return result;
    }
}

    public static void main(String[] args) {
        Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9)
                .reduce((a, b) -> {
                    System.out.println(String.format("%s: %d + %d = %d",
                            Thread.currentThread().getName(), a, b, a + b));
                    return a + b;
                })
                .ifPresent(System.out::println);
    }

main: 1 + 2 = 3
main: 3 + 3 = 6
main: 6 + 4 = 10
main: 10 + 5 = 15
main: 15 + 6 = 21
main: 21 + 7 = 28
main: 28 + 8 = 36
main: 36 + 9 = 45
45

    public static void main(String[] args) {
        Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9)
                .parallel()
                .reduce((a, b) -> {
                    System.out.println(String.format("%s: %d + %d = %d",
                            Thread.currentThread().getName(), a, b, a + b));
                    return a + b;
                })
                .ifPresent(System.out::println);
    }

ForkJoinPool.commonPool-worker-1: 3 + 4 = 7
ForkJoinPool.commonPool-worker-5: 1 + 2 = 3
ForkJoinPool.commonPool-worker-2: 8 + 9 = 17
ForkJoinPool.commonPool-worker-4: 5 + 6 = 11
ForkJoinPool.commonPool-worker-5: 3 + 7 = 10
ForkJoinPool.commonPool-worker-2: 7 + 17 = 24
ForkJoinPool.commonPool-worker-2: 11 + 24 = 35
ForkJoinPool.commonPool-worker-2: 10 + 35 = 45
45

    // ReferencePipeline抽象类的reduce⽅法
    @Override
    public final Optional<P_OUT> reduce(BinaryOperator<P_OUT> accumulator) {
        // 调⽤evaluate⽅法
        return evaluate(ReduceOps.makeRef(accumulator));
    }
    final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {
        assert getOutputShape() == terminalOp.inputShape();
        if (linkedOrConsumed)
            throw new IllegalStateException(MSG_STREAM_LINKED);
        linkedOrConsumed = true;
        return isParallel() // 调⽤isParallel()判断是否使⽤并⾏模式
                ? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOp
                : terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getO
    }
    @Override
    public final boolean isParallel() {
        // 根据之前在parallel()⽅法设置的那个flag来判断。
        return sourceStage.parallel;
    }

// java.util.stream.ReduceOps.ReduceOp.evaluateParallel
@Override
public <P_IN> R evaluateParallel(PipelineHelper<T> helper,
 Spliterator<P_IN> spliterator) {
 return new ReduceTask<>(this, helper, spliterator).invoke().get();
}

public class Test {
    public static void main(String[] args) {
        System.out.println(String.format("本计算机的核数：%d", Runtime.getRuntime().availableProcessors()));
        // 产⽣100w个随机数(1 ~ 100)，组成列表
        Random random = new Random();
        List<Integer> list = new ArrayList<>(1000_0000);
        for (int i = 0; i < 1000_000; i++) {
            list.add(random.nextInt(100));
        }
        long prevTime = getCurrentTime();
        list.stream().reduce((a, b) -> a + b).ifPresent(System.out::println);
        System.out.println(String.format("单线程计算耗时：%d", getCurrentTime() -prevTime));
        prevTime = getCurrentTime();
        list.stream().parallel().reduce((a, b) -> a + b).ifPresent(System.out::println);
        System.out.println(String.format("多线程计算耗时：%d", getCurrentTime() -prevTime));
    }
 
    private static long getCurrentTime() {
        return System.currentTimeMillis();
    }
 
}

本计算机的核数：8
495156156
单线程计算耗时：223
495156156
多线程计算耗时：95