java.util.concurrent 学习笔记(2) 线程池基础

在什么情况下使用线程池？

（1）.单个任务处理的时间比较短

（2）.将需处理的任务的数量大

使用线程池的好处:

(1)、减少在创建和销毁线程上所花的时间以及系统资源的开销

(2)、如不使用线程池，有可能造成系统创建大量线程而导致消耗完系统内存以及”过度切换”。

1、FixedThreadPool

每当提交一个任务就创建一个工作线程，如果工作线程数量达到线程池初始的最大数，则将提交的任务存入到池队列中。
在线程池空闲时，即线程池中没有可运行任务时，它不会释放工作线程，还会占用一定的系统资源

public class FixedThreadPoolTest {

    public static void main(String[] args) {
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(2);
        for (int i = 0; i < 10; i++) {
            Runnable task = new MyTask();
            System.out.println( " Main add Task : " + i);
            fixedThreadPool.execute(task);
        }
        fixedThreadPool.shutdown();

        System.out.println( "Main end ");    }}class MyTask implements Runnable {    @Override    public void run() {        Random random = new Random();        System.out.println(Thread.currentThread().getName() + " begin ");        for (int i = 0; i < 5; i++) {            System.out.println(Thread.currentThread().getName() + " : " + i);            try {                TimeUnit.SECONDS.sleep(random.nextInt(5));            } catch (InterruptedException ex) {            }        }        System.out.println(Thread.currentThread().getName() + " end ");    }}

 

输出：

   Main add Task : 0 

Main add Task : 1 

Main add Task : 2 

Main add Task : 3

pool-1-thread-1 begin  

Main add Task : 4

pool-1-thread-2 begin 

pool-1-thread-2 : 0

pool-1-thread-1 : 0

pool-1-thread-2 : 1

Main end 

pool-1-thread-1 : 1

pool-1-thread-1 : 2

pool-1-thread-2 : 2

pool-1-thread-1 : 3

pool-1-thread-2 : 3

pool-1-thread-1 : 4

pool-1-thread-1 end 

pool-1-thread-1 begin 

pool-1-thread-1 : 0

pool-1-thread-2 : 4

pool-1-thread-2 end 

pool-1-thread-2 begin 

pool-1-thread-2 : 0

pool-1-thread-1 : 1

pool-1-thread-2 : 1

pool-1-thread-2 : 2

pool-1-thread-1 : 2

pool-1-thread-2 : 3

pool-1-thread-2 : 4

pool-1-thread-1 : 3

pool-1-thread-1 : 4

pool-1-thread-1 end 

pool-1-thread-2 end 

pool-1-thread-2 begin 

pool-1-thread-2 : 0

pool-1-thread-2 : 1

pool-1-thread-2 : 2

pool-1-thread-2 : 3

pool-1-thread-2 : 4

pool-1-thread-2 end    

结论：

(1)、线程池大小为2，加入实际线程数大于2时候，线程池也不会阻塞的，也会加进去，只是不马上执行而已。

(2)、线程池的线程可以复用，一直使用两个线程，而不是等线程结束以后，开始新的线程。

(3)、excute() 不会阻塞主线程，即excute() 是异步的。

(4)、线程池必须使用shutdown来显式关闭，否则主线程就无法退出。shutdown也不会阻塞主线程，即shutdown是异步的。

2、SingleThreadExecutor

单线程化的Executor，即只创建唯一的工作者线程来执行任务，如果这个线程异常结束，会有另一个取代它，保证顺序执行(我觉得这点是它的特色)。

单工作线程最大的特点是可保证顺序地执行各个任务，并且在任意给定的时间不会有多个线程是活动的。

        ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 5; i++) {
            Runnable task =  new MyTask();
            System.out.println( " Main add Task : " + i);
            singleThreadExecutor.execute(task);
        }
        singleThreadExecutor.shutdown();
        System.out.println( "Main end ");

输出：

 Main add Task : 0
 Main add Task : 1
 Main add Task : 2
 Main add Task : 3
 Main add Task : 4
pool-1-thread-1 begin 
pool-1-thread-1 : 0
Main end 
pool-1-thread-1 : 1
pool-1-thread-1 : 2
pool-1-thread-1 : 3
pool-1-thread-1 : 4
pool-1-thread-1 end 
pool-1-thread-1 begin 
pool-1-thread-1 : 0
pool-1-thread-1 : 1
pool-1-thread-1 : 2
pool-1-thread-1 : 3
pool-1-thread-1 : 4
pool-1-thread-1 end 
pool-1-thread-1 begin 
pool-1-thread-1 : 0
pool-1-thread-1 : 1
pool-1-thread-1 : 2
pool-1-thread-1 : 3
pool-1-thread-1 : 4
pool-1-thread-1 end 
pool-1-thread-1 begin 
pool-1-thread-1 : 0
pool-1-thread-1 : 1
pool-1-thread-1 : 2
pool-1-thread-1 : 3
pool-1-thread-1 : 4
pool-1-thread-1 end 
pool-1-thread-1 begin 
pool-1-thread-1 : 0
pool-1-thread-1 : 1
pool-1-thread-1 : 2
pool-1-thread-1 : 3
pool-1-thread-1 : 4
pool-1-thread-1 end 

结论：

SingleThreadExecutor 也就是 newFixedThreadPool(1) 而已。

3、CachedThreadPool

(1). 工作线程的创建数量几乎没有限制(其实也有限制的,数目为Interger.MAX_VALUE), 这样可灵活的往线程池中添加线程。

(2).  如果长时间没有往线程池中提交任务，即如果工作线程空闲了指定的时间(默认为1分钟)，则该工作线程将自动终止。终止后，如果你又提交了新的任务，则线程池重新创建一个工作线程。

在使用CachedThreadPool时，一定要注意控制任务的数量，否则，由于大量线程同时运行，很有会造成系统瘫痪

        ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
        for (int i = 0; i < 5; i++) {
            Runnable task =  new MyTask();
            System.out.println( " Main add Task : " + i);
            cachedThreadPool.execute(task);
        }
        cachedThreadPool.shutdown();
        System.out.println( "Main end ");

输出：

 Main add Task : 0
 Main add Task : 1
 Main add Task : 2
 Main add Task : 3
pool-1-thread-1 begin 
pool-1-thread-1 : 0
pool-1-thread-2 begin 
 Main add Task : 4
pool-1-thread-3 begin 
pool-1-thread-3 : 0
pool-1-thread-4 begin 
pool-1-thread-4 : 0
pool-1-thread-2 : 0
pool-1-thread-2 : 1
pool-1-thread-5 begin 
pool-1-thread-5 : 0
Main end 
pool-1-thread-1 : 1
pool-1-thread-2 : 2
pool-1-thread-3 : 1
pool-1-thread-5 : 1
pool-1-thread-4 : 1
pool-1-thread-1 : 2
pool-1-thread-4 : 2
pool-1-thread-1 : 3
pool-1-thread-3 : 2
pool-1-thread-2 : 3
pool-1-thread-5 : 2
pool-1-thread-2 : 4
pool-1-thread-3 : 3
pool-1-thread-3 : 4
pool-1-thread-2 end 
pool-1-thread-5 : 3
pool-1-thread-4 : 3
pool-1-thread-4 : 4
pool-1-thread-3 end 
pool-1-thread-1 : 4
pool-1-thread-4 end 
pool-1-thread-5 : 4
pool-1-thread-1 end 
pool-1-thread-5 end 

结论：

(1)、加入线程后马上执行，不会阻塞的，加多少就运行多少。 

(3)、excute() 不会阻塞主线程，即excute() 是异步的。

(4)、线程池必须使用shutdown来显式关闭，否则主线程就无法退出。shutdown也不会阻塞主线程，即shutdown是异步的。

4、ScheduledThreadPool、SingleScheduledThreadPool

创建一个定长的线程池，而且支持定时的以及周期性的任务执行，类似于Timer

        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(2);
        for (int i = 0; i < 5; i++) {
            Runnable task =  new MyTask();
            System.out.println( " Main add Task : " + i);
            scheduledThreadPool.schedule(task, 1, TimeUnit.SECONDS);
        }
        scheduledThreadPool.shutdown();
       //  scheduledThreadPool.scheduleAtFixedRate(task, 1, 1, TimeUnit.HOURS);

            //scheduledThreadPool.scheduleWithFixedDelay(task, 1, 1, TimeUnit.DAYS); System.out.println( "Main end ");

较好理解。

5、Callable 和 Future

class CallableTask implements Callable {

    public CallableTask() {
    }

    @Override
    public Integer call() throws Exception {
        TimeUnit.SECONDS.sleep(new Random().nextInt(2));
        return new Random().nextInt();
    }
}

public class CallableTest {

    public static void main(String[] args) {
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);
        List futures = new ArrayList();
        for (int i = 0; i < 10; i++) {
            Callable task = new CallableTask();
            Future future = fixedThreadPool.submit(task);
            futures.add(future);
        }
        for (Future future : futures) {
            try {
                if (future.isDone()) {
                    System.out.println("Done: " + future.get());
                } else {
                    future.cancel(true);
                    System.out.println("Canceled: " + future.isCancelled());
                }
            } catch (InterruptedException | ExecutionException ex) {
                ex.printStackTrace();
            }
        }
        fixedThreadPool.shutdown();
        System.out.println("Main end ");
    }
}

 

输出：

Done: 1282190456

Done: -729572913Done: 2080564117Canceled: trueDone: 1361725593Canceled: trueDone: 2038751796Canceled: trueCanceled: trueDone: -1923347889Main end 

结论：

1、submit方法提交 Callable的任务，通过Future 获得结果

2、Future 的get() 方法不马上返回结果，而是要在完成之后返回，

3、Future 的 cancel()方法 取消任务

4、Future 的 isCancel()方法  isDone() 方法返回状态。

6、CompletionService

 public class CompletionServiceTest {

    public static void main(String[] args) {
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);
        CompletionService completionService = new ExecutorCompletionService(fixedThreadPool);
        for (int i = 0; i < 10; i++) {
            Callable task = new CallableTask2();
            completionService.submit(task);
        }
        for (int i = 0; i < 10; i++) {
            try {
                Future take = completionService.take();
                System.out.println(" end: " + take.get());
            } catch (InterruptedException | ExecutionException ex) {
                ex.printStackTrace();
            }
        }
        fixedThreadPool.shutdown();
        System.out.println("Main end ");
    }
}

输出：

 end: -1156022272 

end: -466547416 

end: 1348425806 

end: 558625149 

end: 1573057831 

end: 156442554 

end: 309780892 

end: -333067071 

end: 906757921 

end: -2083965743 

Main end 

结论：

CompletionService接口有两个重要方法：submit()和take()。submit用于提交一个runnable或者callable，一般会提交给一个线程池处理；而take就是取出已经执行完毕runnable或者callable实例的Future对象，如果没有满足要求的，就等待了; 方法poll()与take类似，只是不会等待，如果没有满足要求，就返回null对象。

public class InvokeAnyTest {

    public static void main(String[] args) {

        ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(5);
        List tasks = new LinkedList();
        for (int i = 0; i < 10; i++) {
            CallableTask callableTask = new CallableTask(i + "");
            tasks.add(callableTask);
        }
        try {
            Integer result = newFixedThreadPool.invokeAny(tasks);
            System.out.println("first result at " + System.currentTimeMillis() + ":" + result);
        } catch (InterruptedException | ExecutionException ex) {
            System.out.println("invokeAny exception at " + System.currentTimeMillis() + ":" + ex.getMessage());
        }

        newFixedThreadPool.shutdown();
    }
}

class CallableTask implements Callable {

    String name;

    public CallableTask(String name) {
        this.name = name;
    }

    @Override
    public Integer call() {
        try {
            TimeUnit.SECONDS.sleep(new Random().nextInt(5));
        } catch (InterruptedException ex) {
            System.out.println(this.name + " exception at " + System.currentTimeMillis() + ":" + ex.getMessage());
        }
        int nextInt = new Random().nextInt(1000);
        System.out.println(this.name + " return at " + System.currentTimeMillis() + ":" + nextInt);
        return nextInt;
    }

 

输出：

2 return at 1411354042128:7814 

return at 1411354042128:1833 

return at 1411354042128:2307 

return at 1411354042129:795

first result at 1411354042129:781

6 exception at 1411354042130:sleep interrupted

0 exception at 1411354042130:sleep interrupted

5 exception at 1411354042130:sleep interrupted

0 return at 1411354042130:435

8 exception at 1411354042130:sleep interrupted

8 return at 1411354042130:939

1 exception at 1411354042130:sleep interrupted

6 return at 1411354042130:428

1 return at 1411354042130:519

5 return at 1411354042130:807

结论：

1、只返回最早那一个结果

2、其余没完成的线程中断，结果忽略。

8. InvokeAll

public class InvokeAllTest {

    public static void main(String[] args) {

        ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(5);
        List tasks = new LinkedList();
        for (int i = 0; i < 10; i++) {
            CallableTask2 callableTask = new CallableTask2(i + "");
            tasks.add(callableTask);
        }
        try {
            List> listReslut = newFixedThreadPool.invokeAll(tasks);
            for (Future result : listReslut) {
                System.out.println(" result :" + result.get());

            }
        } catch (InterruptedException | ExecutionException ex) {
            System.out.println("invokeAny exception at " + System.currentTimeMillis() + ":" + ex.getMessage());
        }
        newFixedThreadPool.shutdown();
    }
}

class CallableTask2 implements Callable {

    String name;

    public CallableTask2(String name) {
        this.name = name;
    }

    @Override
    public Integer call() {
        try {
            TimeUnit.SECONDS.sleep(new Random().nextInt(5));
        } catch (InterruptedException ex) {
            System.out.println(this.name + " exception at " + System.currentTimeMillis() + ":" + ex.getMessage());
        }
        int nextInt = new Random().nextInt(1000);
        System.out.println(this.name + " return at " + System.currentTimeMillis() + ":" + nextInt);
        return nextInt;
    }
}

输出：

1 return at 1411354530402:993
3 return at 1411354531402:706
4 return at 1411354533402:998
0 return at 1411354534402:116
2 return at 1411354534402:432
6 return at 1411354534402:298
5 return at 1411354534402:206
8 return at 1411354534402:812
7 return at 1411354535402:57
9 return at 1411354537402:957
 result :116
 result :993
 result :432
 result :706
 result :998
 result :206
 result :298
 result :57
 result :812
 result :957

结论：全部任务完成再返回。

7、FutureTask

FutureTask 实现 Runnable 和Future 接口，没有实现Callable接口。

public class FutureTask implements RunnableFuture

public interface RunnableFuture extends Runnable, Future {

    void run();
}

public interface Future {
 
    boolean cancel(boolean mayInterruptIfRunning);
 
    boolean isCancelled();
 
    boolean isDone();
 
    V get() throws InterruptedException, ExecutionException;
  
    V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
}

例子

public class FutureTaskTest {

    public static void main(String[] args) throws InterruptedException, ExecutionException {

        ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(2);
        
        //run FutureTask by ThreadPool 
        List futureTasks = new LinkedList();
        for (int i = 0; i < 5; i++) {
            CallableTask callableTask = new CallableTask(i + "--pool");
            FutureTask ft = new FutureTask(callableTask);
            futureTasks.add(ft);
            newFixedThreadPool.execute(ft);
        }
        for (FutureTask ft : futureTasks) {
            System.out.println("By ThreadPool : " + ft.get());
        }

          //run FutureTask by standalone Thread
        for (int i = 0; i < 5; i++) {
            CallableTask callableTask = new CallableTask(i + "--alone");
            FutureTask ft = new FutureTask(callableTask);
            new Thread(ft).start();
        }
        for (FutureTask ft : futureTasks) {
            System.out.println("Standalone Thread : " + ft.get());
        }

        newFixedThreadPool.shutdown();
    }
}

 

输出：

0--pool return at 1411356206910:86
By ThreadPool : 86
1--pool return at 1411356208910:128
By ThreadPool : 128
2--pool return at 1411356208911:109
By ThreadPool : 109
3--pool return at 1411356211910:258
By ThreadPool : 258
4--pool return at 1411356212911:792
By ThreadPool : 792
1--alone return at 1411356212912:556
Standalone Thread : 86
Standalone Thread : 128
Standalone Thread : 109
Standalone Thread : 258
Standalone Thread : 792
4--alone return at 1411356214913:404
0--alone return at 1411356215913:949
3--alone return at 1411356215914:507
2--alone return at 1411356216913:769

8、自定义线程池的分析和使用

我们可以通过ThreadPoolExecutor来创建一个线程池。

new  ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, milliseconds,runnableTaskQueue, handler);

创建一个线程池需要输入几个参数：

corePoolSize（线程池的基本大小）：当提交一个任务到线程池时，线程池会创建一个线程来执行任务，即使其他空闲的基本线程能够执行新任务也会创建线程，等到需要执行的任务数大于线程池基本大小时就不再创建。如果调用了线程池的prestartAllCoreThreads方法，线程池会提前创建并启动所有基本线程。

runnableTaskQueue（任务队列）：用于保存等待执行的任务的阻塞队列。 可以选择以下几个阻塞队列。

            ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按 FIFO（先进先出）原则对元素进行排序。

            LinkedBlockingQueue：一个基于链表结构的阻塞队列，此队列按FIFO （先进先出） 排序元素，吞吐量通常要高于ArrayBlockingQueue。静态工厂方法Executors.newFixedThreadPool()使用了这个队列。

            SynchronousQueue：一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue，静态工厂方法Executors.newCachedThreadPool使用了这个队列。

            PriorityBlockingQueue：一个具有优先级的无限阻塞队列。

maximumPoolSize（线程池最大大小）：线程池允许创建的最大线程数。如果队列满了，并且已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务。值得注意的是如果使用了无界的任务队列这个参数就没什么效果。

threadFactory：用于设置创建线程的工厂，可以通过线程工厂给每个创建出来的线程设置更有意义的名字。

RejectedExecutionHandler（饱和策略）：当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是                     AbortPolicy，表示无法处理新任务时抛出异常。以下是JDK1.5提供的四种策略。

             AbortPolicy：直接抛出异常。

             CallerRunsPolicy：只用调用者所在线程来运行任务。

             DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务。

             DiscardPolicy：不处理，丢弃掉。

             当然也可以根据应用场景需要来实现RejectedExecutionHandler接口自定义策略。如记录日志或持久化不能处理的任务。

keepAliveTime（线程活动保持时间）：线程池的工作线程空闲后，保持存活的时间。所以如果任务很多，并且每个任务执行的时间比较短，可以调大这个时间，提高线程的利用率。

TimeUnit（线程活动保持时间的单位）：可选的单位有DAYS，HOURS，MINUTES，MILLISECONDS，MICROSECONDS,NANOSECONDS

当提交一个新任务到线程池时，线程池的处理流程如下：

1.   首先线程池判断基本线程池是否已满？没满，创建一个工作线程来执行任务。满了，则进入下个流程。

2.   其次线程池判断工作队列是否已满？没满，则将新提交的任务存储在工作队列里。满了，则进入下个流程。

3.   最后线程池判断整个线程池是否已满？没满，则创建一个新的工作线程来执行任务，满了，则交给饱和策略来处理这个任务