一篇文章带你搞定线程池的自定义创建和扩展-蒲公英云

一篇文章带你搞定线程池的自定义创建和扩展

た入场券 2022-12-25 06:00 295阅读 0赞

文章目录

- 一、自定义线程创建：ThreadFactory
- 二、扩展线程池

一、自定义线程创建：ThreadFactory

看了那么多有关线程池的介绍，不知道大家有没有思考过一个基本的问题：线程池中的线程是从哪里来的呢？

之前我们介绍过，线程池的主要作用是为了线程复用，也就是避免了线程的频繁创建。但是，最开始的那些线程从何而来呢？答案就是ThreadFactory。

ThreadFactory是一个接口，它只有一个用来创建线程的方法。

Thread newThread(Runnable r);

当线程池需要新建线程时，就会调用这个方法。

自定义线程池可以帮助我们做不少事。比如，我们可以跟踪线程池究竟在何时创建了多少线程，也可以自定义线程的名称、组以及优先级等信息，甚至可以任性地将所有的线程设置为守护线程。总之，使用自定义线程池可以让我们更加自由地设置线程池中所有线程的状态。下面的案例使用自定义的ThreadFactory，一方面记录了线程的创建，另一方面将所有的线程都设置为守护线程，这样，当主线程退出后，将会强制销毁线程池。

import java.util.concurrent.*;
public class MyThreadFactory { 
    public static class MyTask implements Runnable { 
        @Override
        public void run() { 
            System.out.println(System.currentTimeMillis() + "thread id:" + Thread.currentThread().getId());
            try { 
                Thread.sleep(100);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    }
    public static void main(String[] args) throws InterruptedException { 
        MyTask task = new MyTask();
        ExecutorService es = new ThreadPoolExecutor(5, 5,
                0L, TimeUnit.MILLISECONDS,
                //使用直接提交的队列
                new SynchronousQueue<Runnable>(),
                new ThreadFactory() { 
                    @Override
                    public Thread newThread(Runnable r) { 
                        Thread t = new Thread(r);
                        //设置守护线程
                        t.setDaemon(true);
                        System.out.println("create " + t);
                        return t;
                    }
                });
        for (int i = 0; i < 5; i++) { 
            es.submit(task);
        }
        Thread.sleep(2000);
    }
}

在这里插入图片描述

二、扩展线程池

虽然JDK已经帮我们实现了这个稳定的高性能线程池，但如果我们需要对这个线程池做一些扩展，比如，监控每个任务执行的开始时间和结束时间，或者其他一些自定义的增强功能，这时候应该怎么办呢？

一个好消息是：ThreadPoolExecutor是一个可以扩展的线程池。它提供了beforeExecute()、afterExecute()和terminated()三个接口用来对线程池进行控制。

以beforeExecute()、afterExecute()两个接口为例，它们在ThreadPoolExecutor.Worker.runTask()方法内部提供了这样的实现：

在这里插入图片描述
ThreadPoolExecutor.Worker是ThreadPoolExecutor的内部类，它是一个实现了Runnable接口的类。ThreadPoolExecutor线程池中的工作线程也正是Worker实例。Worker.run()方法会调用上述ThreadPoolExecutor.runWorker(Worker w)实现每一个工作线程的固有工作。

在默认的ThreadPoolExecutor实现中，提供了空的beforeExecute()和afterExecute()两个接口实现。在实际应用中，可以对其进行扩展来实现对线程池运行状态的跟踪，输出一些有用的调试信息，以帮助系统故障诊断，这对于多线程程序错误排查是很有帮助的。下面演示了对线程池的扩展，在这个扩展中，我们将记录每一个任务的执行日志。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ExtThreadPool { 
    public static class MyTask implements Runnable { 
        public String name;
        public MyTask(String name) { 
            this.name = name;
        }
        @Override
        public void run() { 
            System.out.println("正在执行" + ":Thread id" + Thread.currentThread().getId() + ", Task name:" + this.name);
        }
    }
    public static void main(String[] args) throws InterruptedException { 
        ExecutorService es = new ThreadPoolExecutor(5, 5, 0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingDeque<Runnable>()) { 
                    @Override
                    protected void beforeExecute(Thread t, Runnable r) { 
                        System.out.println("准备执行：" + ((MyTask) r).name);
                    }
                    @Override
                    protected void afterExecute(Runnable r, Throwable t) { 
                        System.out.println("执行完成：" + ((MyTask) r).name);
                    }
                    @Override
                    protected void terminated() { 
                        System.out.println("线程池退出！");
                    }
        };
        for (int i = 0; i < 5; i++) { 
            MyTask task = new MyTask("TASK-GEYM-" + i);
            es.execute(task);
            Thread.sleep(10);
        }
        es.shutdown();
    }
}

在提交完成后，调用shutdown()方法关闭线程池。这是一个比较安全的方法，如果当前正有线程在执行，shutdown()方法并不会立即暴力地终止所有任务，它会等待所有任务执行完成后，再关闭线程池，但它并不会等待所有线程执行完成后再返回，因此，可以简单地理解成shutdown()方法只是发送了一个关闭信号而已。但在shutdown()方法执行后，这个线程池就不能再接受其他新的任务了。

在这里插入图片描述