Java 线程池

2021-05-06 13:27

阅读：607

标签：tom 停止资源决定 info nbsp lock 检测 print

线程

Java多线程，皆始于Thread。Thread是多线程的根，每一个线程的开启都始于Thread的start()方法。

技术图片

Runnable

看一个例子：

Thread thread = new Thread(new Runnable() {
    @Override
    public void run() {
        System.out.println("thread begin...");
        try {
            Thread.sleep(1000 * 30);
        } catch (Exception e) {

        }
        System.out.println("thread end");
    }
});


thread.start();

try {
    thread.join();
} catch (Exception e) {

}

System.out.println("main done");

new 一个 Thread，然后调用其 start() 方法，会启动一个线程并使线程进入了就绪状态，当分配到时间片后就可以开始运行了。 start() 会执行线程的相应准备工作，然后自动执行 run() 方法的内容。如果直接执行Thread的 run() 方法，会把 run 方法当成一个 main 线程下的普通方法去执行，并不会在某个线程中执行它。

调用 start 方法方可启动线程并使线程进入就绪状态，而 run 方法只是 thread 的一个普通方法调用，还是在主线程里执行，此外：

start方法用synchronized修饰，为同步方法；
虽然为同步方法，但不能避免多次调用问题，用threadStatus来记录线程状态，如果线程被多次start会抛出异常；threadStatus的状态由JVM控制。
使用Runnable时，主线程无法捕获子线程中的异常状态。线程的异常，应在线程内部解决。

Callable

如果想捕获线程的返回值或异常，可以实现Callable接口

FutureTask futureTask = new FutureTask(new Callable() {
    @Override
    public String call() throws Exception {
        return "hello world!";
    }
});

Thread thread = new Thread(futureTask);

thread.start();

try {
    String result = futureTask.get();
    System.out.println(result);
} catch (Exception e) {

}

System.out.println("main done");

中止线程

如果用Thread.stop()方法中止一个正在运行的线程：

public class MyThread {

    public static void main(String[] args) throws InterruptedException {
        StopThread thread = new StopThread();
        thread.start();

        // 休眠1秒，确保线程进入运行
        Thread.sleep(1000);

        // 暂停线程
        thread.stop();
        //thread.interrupt();

        // 确保线程已经销毁
        while (thread.isAlive()) { }

        thread.print();
    }

    private static class StopThread extends Thread {

        private int x = 0;
        private int y = 0;

        @Override
        public void run() {
            // 这是一个同步原子操作
            synchronized (this) {
                ++x;
                try {
                    // 休眠3秒,模拟耗时操作
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                ++y;
            }
        }

        public void print() {
            System.out.println("x=" + x + " y=" + y);
        }
    }
}

运行结果：

x=1 y=0

线程没有抛出异常，如果把stop改成interrupt，那么运行结果变为

x=1 y=1
java.lang.InterruptedException: sleep interrupted
    at java.lang.Thread.sleep(Native Method)
    at com.meitu.helloworld.MyThread$StopThread.run(MyThread.java:34)

stop() 方法事实上已被废弃，它对线程的强制中断是不可预期的。

interrupt() 方法是一个比较温柔的做法，它更类似一个标志位。它不能中断线程，而是「通知线程应该中断了」，具体到底中断还是继续运行，应该由被通知的线程自己处理：

具体来说，当对一个线程，调用 interrupt() 时，

如果线程处于被阻塞状态（例如处于sleep, wait, join 等状态），那么线程将立即退出被阻塞状态，并抛出一个InterruptedException异常，仅此而已。
如果线程处于正常活动状态，那么会将该线程的中断标志设置为 true，仅此而已。被设置中断标志的线程将继续正常运行，不受影响。

可见，interrupt() 并不能真正的中断线程，需要被调用的线程自己进行配合才行。也就是说，一个线程如果有被中断的需求，那么就可以这样做。

在正常运行任务时，经常检查本线程的中断标志位，如果被设置了中断标志就自行停止线程；
在调用阻塞方法时正确处理InterruptedException异常；

最后，总结下创建Thread的三种方法：

继承Thread类，重写run方法；
实现Runnable的run方法；
实现Callable的call方法；

线程池

技术图片

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler);

参数：

corePoolSize : 核心线程数，一旦创建将不会再释放；
maximumPoolSize : 最大线程数，如果最大线程数等于核心线程数，则无法创建非核心线程；如果非核心线程处于空闲时，超过设置的空闲时间，则将被回收，释放占用的资源。
keepAliveTime : 也就是当线程空闲时，所允许保存的最大时间，超过这个时间，线程将被释放销毁，但只针对于非核心线程。
unit : 时间单位，TimeUnit.SECONDS等。
workQueue : 任务队列，用于保存等待执行的任务的阻塞队列，可以选择以下几个阻塞队列。

ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，必须设置容量。此队列按 FIFO（先进先出）原则对元素进行排序。
LinkedBlockingQueue：一个基于链表结构的阻塞队列，可以设置容量，此队列按FIFO （先进先出）排序元素，吞吐量通常要高于ArrayBlockingQueue。
SynchronousQueue：一个不存储元素的阻塞队列。每个插入offer操作必须等到另一个线程调用移除poll操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue。
PriorityBlockingQueue：一个具有优先级的无限阻塞队列。

threadFactory : 线程工厂，用于创建线程。
handler : 拒绝策略，当线程边界和队列容量已经达到最大时，用于处理阻塞时的程序：

AbortPolicy：默认策略，抛出异常RejectedExecutionException拒绝提交任务；
CallerRunsPolicy：由调用execute方法提交任务的线程来执行这个任务；
DiscardPolicy：直接抛弃任务，不做任何处理；
DiscardOldestPolicy：去除任务队列中的第一个任务，重新提交；

处理任务

提交任务有2个方法：

execute(Runnable)，无返回值；
submit(Callable)，有返回值；

看一个例子：

public class ThreadTest {

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

        final AtomicInteger mThreadNum = new AtomicInteger(1);

        ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4,
                10, TimeUnit.SECONDS,
                new ArrayBlockingQueue(2),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(@NotNull Runnable r) {
                        Thread t = new Thread(r, "my-thread-" + mThreadNum.getAndIncrement());
                        System.out.println(t.getName() + " has been created");
                        return t;
                    }
                },
                new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
                        System.err.println(r.toString() + " rejected");
                    }
                });

        executor.prestartAllCoreThreads(); // 预启动所有核心线程

        for (int i = 1; i ) {
            MyTask task = new MyTask(String.valueOf(i));
            executor.execute(task);
        }

        executor.shutdown();
    }

    static class MyTask implements Runnable {
        private String name;

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

        @Override
        public void run() {
            try {
                System.out.println(this.toString() + " is running!");
                Thread.sleep(3000); //让任务执行慢点
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        
        @Override
        public String toString() {
            return "MyTask [name=" + name + "]";
        }
    }
}

运行结果：

my-thread-1 has been created
my-thread-2 has been created
my-thread-3 has been created
MyTask [name=1] is running!
my-thread-4 has been created
MyTask [name=3] is running!
MyTask [name=2] is running!
MyTask [name=5] is running!
MyTask [name=7] rejected
MyTask [name=8] rejected
MyTask [name=9] rejected
MyTask [name=10] rejected
MyTask [name=4] is running!
MyTask [name=6] is running!

其中线程线程1-4先占满了核心线程和最大线程数量，然后4、5线程进入等待队列，7-10线程被直接忽略拒绝执行，等1-4线程中有线程执行完后通知4、5线程继续执行。

当在execute(Runnable)方法中提交新任务并且少于corePoolSize线程正在运行时，即使其他工作线程处于空闲状态，也会创建一个新线程来处理该请求。如果有多于corePoolSize但小于maximumPoolSize线程正在运行，则仅当队列已满时才会创建新线程。通过设置corePoolSize和maximumPoolSize相同，您可以创建一个固定大小的线程池。通过将maximumPoolSize设置为基本上无界的值，例如Integer.MAX_VALUE，您可以允许池容纳任意数量的并发任务。

技术图片

如果运行的线程少于 corePoolSize，则 Executor 始终首选添加新的线程，而不进行排队。（即如果当前运行的线程小于corePoolSize，则任务根本不会添加到workQueue中）
如果运行的线程等于或多于 corePoolSize，则 Executor 始终首选将请求加入工作队列，而不添加新的线程。
如果无法将请求加入workQueue（但是队列已满），则创建新的线程，除非创建此线程超出 maximumPoolSize，如果超过，在这种情况下，新的任务将被拒绝。

预定义线程池

FixedThreadPool

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue());
    }

corePoolSize与maximumPoolSize相等，即其线程全为核心线程，是一个固定大小的线程池。

SingleThreadExecutor

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue()));
    }

看起来很像 newFixedThreadPool(1)，但多了一层 FinalizableDelegatedExecutorService 包装，看下它的作用：

ExecutorService fixedThreadPool = Executors.newFixedThreadPool(1);
((ThreadPoolExecutor) fixedThreadPool).setMaximumPoolSize(3);
System.out.println(((ThreadPoolExecutor) fixedThreadPool).getMaximumPoolSize());    // 3

ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
((ThreadPoolExecutor) singleThreadExecutor).setMaximumPoolSize(3);  // 运行时异常 java.lang.ClassCastException
//System.out.println(((ThreadPoolExecutor) singleThreadExecutor).getMaximumPoolSize());

可见，SingleThreadExecutor被包装后，无法成功向下转型。因此，SingleThreadExecutor被定以后，无法修改，做到了真正的Single。

newCachedThreadPool

 public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue());
    }

corePoolSize = 0，maximumPoolSize = Integer.MAX_VALUE，即其线程全为非核心线程，空闲超时会被释放。

ScheduledThreadPool

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }

newScheduledThreadPool调用的是ScheduledThreadPoolExecutor的构造方法，

而ScheduledThreadPoolExecutor继承了ThreadPoolExecutor，构造是还是调用了其父类的构造方法。

public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
    }

关闭线程池

线程池自动关闭的两个条件：1、线程池的引用不可达；2、线程池中没有线程；

如果核心线程不为0，由于没有超时策略，所以并不会自动关闭。

当shutdown一个线程池后，继续提交任务，会执行拒绝策略；

public static void main(String[] args) {
    ThreadPoolExecutor executor = new ThreadPoolExecutor(4, 4, 10, TimeUnit.SECONDS, new LinkedBlockingQueue());
    executor.execute(() -> System.out.println("before shutdown"));
    executor.shutdown();
    executor.execute(() -> System.out.println("after shutdown"));
}

shutdown一个线程池后，等待队列的任务仍会被继续执行，但如果用shutdownNow()方法，则不执行队列中的任务；

shutdown和shutdownNow对正在执行的任务的影响是怎样的呢？

public class InteruptTest {

    public static void main(String[] args) throws InterruptedException {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 10, TimeUnit.SECONDS, new LinkedBlockingQueue());
        executor.execute(new Task("0"));
        Thread.sleep(1);
        executor.shutdownNow();
        System.out.println("executor has been shutdown");
    }

    static class Task implements Runnable {
        String name;

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

        @Override
        public void run() {

            for (int i = 1; i ) {
                Thread.yield();
                System.out.println("task " + name + " is running, round " + i);
            }

        }
    }
}

运行结果：

task 0 is running, round 1
task 0 is running, round 2
task 0 is running, round 3
task 0 is running, round 4
task 0 is running, round 5
task 0 is running, round 6
task 0 is running, round 7
task 0 is running, round 8
task 0 is running, round 9
task 0 is running, round 10
task 0 is running, round 11
task 0 is running, round 12
task 0 is running, round 13
task 0 is running, round 14
task 0 is running, round 15
task 0 is running, round 16
task 0 is running, round 17
task 0 is running, round 18
task 0 is running, round 19
executor has been shutdown
task 0 is running, round 20

shutdownNow 会将正在执行任务的Thread.interrupted 置为true，如果线程检测了该状态，可以决定要不要停止运行。

Java 线程池

标签：tom 停止资源决定 info nbsp lock 检测 print

原文地址：https://www.cnblogs.com/chenny7/p/13187826.html

上一篇：922. 按奇偶排序数组 II『简单』

下一篇：详细介绍去一年在 PyPI 上下载次数最多的 Python 包

文章来自：搜素材网的编程语言模块，转载请注明文章出处。
文章标题：Java 线程池
文章链接：http://soscw.com/index.php/essay/83224.html

亲，登录后才可以留言！

Java 线程池

线程

Runnable

Callable

中止线程

线程池

处理任务

预定义线程池

FixedThreadPool

SingleThreadExecutor

newCachedThreadPool

ScheduledThreadPool

关闭线程池

评论

热门文章

推荐文章

最新文章

置顶文章