实验讲解-线程池停止执行任务 Executor.execute & shutdown & awaitTermination & shutdownNow()

1 Executor.execute

package com.zs.thread;
public class TestVolatile {
    public static void main(String[] args) {
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);

        for (int i = 0; i < 5; i++) {
            final int index = i;
            fixedThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
                        System.out.println("运行时间: " + sdf.format(new Date()) + " " + index);
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        fixedThreadPool.shutdown();
    }
}


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

public interface Executor {
在将来的某个时间执行给定的可运行的任务。该可运行的任务可以在新线程、池线程或调用线程中执行，由Executor实现决定。 
参数： command–可运行的任务 
投掷： RejectedExecutionException–如果无法接受执行此任务 
	  NullPointerException–如果命令为空
void execute(Runnable command);
}
1
2
3
4
5
6
7

2 ExecutorService.shutdown()

1、停止接收新的submit的任务；
2、已经提交的任务（包括正在跑的和队列中等待的）,会继续执行完成；
3、等到第2步完成后，才真正停止；

public interface ExecutorService extends Executor {
 void shutdown();
}
1
2
3

实例如下：

public class ThreadPoolExecutorCloseTest {
    public static void main(String[] args) {
        ExecutorService executorService = new ThreadPoolExecutor(10,
                20,
                30,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1),
                Thread::new,
                new ThreadPoolExecutor.AbortPolicy());

        IntStream.range(0, 20).boxed().forEach(i -> {
            executorService.execute(() -> {
                try {
                    TimeUnit.SECONDS.sleep(10);
                    System.out.println(Thread.currentThread().getName() + " [ " + i + " ] finish done.");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        });
        /**
         * 1、停止接收新的submit的任务；
         * 2、已经提交的任务（包括正在跑的和队列中等待的）会继续执行完成；
         * 3、等到第2步完成后，才真正停止；
         */
        executorService.shutdown();
        System.out.println("==============over==============");
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

线程真正停止了！

==============over==============
Thread-15 [ 16 ] finish done.
Thread-9 [ 9 ] finish done.
Thread-16 [ 17 ] finish done.
Thread-7 [ 7 ] finish done.
Thread-12 [ 13 ] finish done.
Thread-5 [ 5 ] finish done.
Thread-4 [ 4 ] finish done.
Thread-3 [ 3 ] finish done.
Thread-1 [ 1 ] finish done.
Thread-11 [ 12 ] finish done.
Thread-6 [ 6 ] finish done.
Thread-17 [ 18 ] finish done.
Thread-10 [ 11 ] finish done.
Thread-13 [ 14 ] finish done.
Thread-18 [ 19 ] finish done.
Thread-8 [ 8 ] finish done.
Thread-14 [ 15 ] finish done.
Thread-2 [ 2 ] finish done.
Thread-0 [ 0 ] finish done.
Thread-12 [ 10 ] finish done.

Process finished with exit code 0

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

3 ExecutorService.awaitTermination()

当前线程阻塞，直到：
1、等所有已提交的任务（包括正在跑的和队列中等待的）执行完；
2、或者 等超时时间到了（timeout 和 TimeUnit设定的时间）；
3、或者 线程被中断，抛出InterruptedException

然后会监测 ExecutorService 是否已经关闭，返回true（shutdown请求后所有任务执行完毕）或false（已超时）

boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;
1
2
3

示例如下：

public class ThreadPoolExecutorCloseTest {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = new ThreadPoolExecutor(10,
                20,
                30,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1),
                Thread::new,
                new ThreadPoolExecutor.AbortPolicy());

        IntStream.range(0, 20).boxed().forEach(i -> {
            executorService.execute(() -> {
                try {
                    TimeUnit.SECONDS.sleep(10);
                    System.out.println(Thread.currentThread().getName() + " [ " + i + " ] finish done.");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        });
        executorService.shutdown();
        /**
         * 当前线程阻塞，直到：
         * 1、等所有已提交的任务（包括正在跑的和队列中等待的）执行完
         * 2、或者 等超时时间到了（timeout 和 TimeUnit设定的时间）
         * 3、或者 线程被中断，抛出InterruptedException
         * 然后会监测 ExecutorService 是否已经关闭，返回true（shutdown请求后所有任务执行完毕）或false（已超时）
         */
        boolean b = executorService.awaitTermination(30, TimeUnit.SECONDS);
        System.out.println(b);
        System.out.println("==============over==============");
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

线程真正停止！

Thread-16 [ 17 ] finish done.
Thread-0 [ 0 ] finish done.
Thread-14 [ 15 ] finish done.
Thread-13 [ 14 ] finish done.
Thread-7 [ 7 ] finish done.
Thread-3 [ 3 ] finish done.
Thread-2 [ 2 ] finish done.
Thread-5 [ 5 ] finish done.
Thread-1 [ 1 ] finish done.
Thread-17 [ 18 ] finish done.
Thread-11 [ 12 ] finish done.
Thread-12 [ 13 ] finish done.
Thread-10 [ 11 ] finish done.
Thread-18 [ 19 ] finish done.
Thread-9 [ 9 ] finish done.
Thread-6 [ 6 ] finish done.
Thread-8 [ 8 ] finish done.
Thread-15 [ 16 ] finish done.
Thread-4 [ 4 ] finish done.
Thread-13 [ 10 ] finish done.
true
==============over==============

Process finished with exit code 0

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

4 shutdownNow()

停止接收新任务，原来的任务停止执行
1、跟 shutdown() 一样，先停止接收新submit的任务；
2、忽略队列里等待的任务；
3、尝试将正在执行的任务interrupt中断；
4、返回未执行的任务列表 — new ArrayBlockingQueue<>(8)

说明：它试图终止线程的方法是通过调用 Thread.interrupt() 方法来实现的，这种方法的作用有限，如果线程中没有sleep 、wait、Condition、定时锁等应用, interrupt() 方法是无法中断当前的线程的。所以，shutdownNow() 并不代表线程池就一定立即就能退出，它也可能必须要等待所有正在执行的任务都执行完成了才能退出。但是大多数时候是能立即退出的。

public class ThreadPoolExecutorCloseTest {
    public static void main(String[] args) {
        ExecutorService executorService = new ThreadPoolExecutor(10,
                20,
                30,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(8),
                Thread::new,
                new ThreadPoolExecutor.AbortPolicy());

        IntStream.range(0, 20).boxed().forEach(i -> {
            executorService.execute(() -> {
                try {
                    TimeUnit.SECONDS.sleep(10);
                    System.out.println(Thread.currentThread().getName() + " [ " + i + " ] finish done.");
                } catch (InterruptedException e) {
//                    e.printStackTrace();
                }
            });
        });
        shutdownNow(executorService);
        System.out.println("==============over==============");
    }

    private static void shutdownNow(ExecutorService executorService) {
        List<Runnable> runnableList = Lists.newArrayList();
        try {
            runnableList = executorService.shutdownNow();
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println(runnableList.size());
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

8
==============over==============

Process finished with exit code 0
1
2
3
4

shutdown只是将线程池的状态设置为SHUTWDOWN状态，正在执行的任务会继续执行下去，没有被执行的则中断。

而shutdownNow则是将线程池的状态设置为STOP，正在执行的任务则被停止，没被执行任务的则返回。

5 长时间未运行完的线程处理

package com.thread.excutor;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

public class ThreadPoolExecutorLongTimeTest {
    public static void main(String[] args) {
        ExecutorService executorService = new ThreadPoolExecutor(10,
                20,
                30,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(8),
                Thread::new,
                new ThreadPoolExecutor.AbortPolicy());

        IntStream.range(0, 10).boxed().forEach(item -> {
            executorService.submit(() -> {
                while (true) {
                }
            });
        });

        try {
            executorService.shutdown();
            executorService.awaitTermination(5, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println("===============over================");
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

线程池一直活着，无法停止！

package com.thread.excutor;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

public class ThreadPoolExecutorLongTimeTest {
    public static void main(String[] args) {
        ExecutorService executorService = new ThreadPoolExecutor(10,
                20,
                30,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(8),
                target -> {
                    Thread thread = new Thread(target);
                    thread.setDaemon(true);
                    return thread;
                },
                new ThreadPoolExecutor.AbortPolicy());

        IntStream.range(0, 10).boxed().forEach(item -> {
            executorService.submit(() -> {
                while (true) {
                }
            });
        });

        try {
            executorService.shutdown();
            executorService.awaitTermination(5, TimeUnit.SECONDS); // 最多的期待时间
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println("===============over================");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

ok！设置守护线程的手段！