并发编程学习笔记之自定义配置线程池(七)

等待其他资源,可能会产生线程饥饿死锁

在线程池中如果一个任务依赖于其它任务的执行,就可能产生死锁.在一个单线程化的Executor中,提交两个任务,任务二滞留在工作队列中等待第一个任务完成,但是第一个任务不会完成,因为它在等待第二个任务的完成(需要第二个任务执行的结果进行运算),这就会发生死锁.

在一个大的线程池中,如果所有线程执行的任务都阻塞在线程池中,等待着仍然处于同一工作队列中的其它任务,那么会发生同样的问题.这被称作线程饥饿死锁(thread starvation deadlock)

产生死锁的情况: 只要池任务开始了无限期的阻塞,其目的是等待一些资源或条件,此时只有另一个池任务的活动才能使那些条件成立,比如等待返回值.除非你能保证这个池足够大,否则会产生线程饥饿死锁.

池任务等待另一个池任务的结果,可能会发生死锁:

public class ThreadDeadLock {
     ExecutorService exec = Executors.newSingleThreadExecutor();



   public  class Task implements Callable{

        @Override
        public Object call() throws Exception {
            //等待另一个池任务的结果
            Future<String> future1 = exec.submit(new LockTask());
            Future<String> future2 = exec.submit(new LockTask());
            //可能发生死锁
            return future1.get()+future2.get();

        }
    }

    

无论何时,你提交了一个非独立的Executor任务,要明确出现线程饥饿死锁的可能性,并且,在代码或者配置文件以及其他可以配置Executor的地方,任何有关池的大小和配置约束都要写入文档

耗时的任务,设置超时时间

如果你的线程池不够大,又有很多耗时的任务,这会影响服务的响应性.这时候你可以限定任务等待资源的时间,而不是无限制地等下去.

耗时的任务可能会死锁或者响应的很慢

大多数平台类库中的阻塞方法,都有限时和非限时两个版本.例如Blocking.put.如果超时你可以把任务标记为失败,终止或者把他重新返回队列,准备之后执行.这样无论每个任务的最终结果是成功还是失败,都保证了任务会向前发展,这样可以更快地将线程从任务中解放出来.(如果线程池频频被阻塞的任务充满,这同样可能是池太小的一个信号).

定制线程池的大小

不要硬编码线程池的大小

线程池合理的长度取决于未来提交的任务类型和所部属系统的特征.池的长度应该由某种配置机制来提供,或者利用Runtime.availableProcessors(获取你电脑的处理器数量),动态进行计算

线程池过大&过小的坏处

线程池过大: 线程对稀缺的CPU和内存资源的竞争,会导致内存的高使用量.线程间切换也会消耗资源

线程池过小:由于存在很多可用的处理器资源没用,会对吞吐量造成损失

制定线程池大小依据的内容

正确的定制线程池的长度,需要理解你的计算环境、资源预算和任务的自身特性.

部署系统中安装了多少个CPU?多少内存?任务主要执行的是计算、I/O还是一些混合操作?它们是否需要像JDBC Connection 这样的稀缺资源?

如果你有不同类别的任务,它们拥有差别很大行为,那么请考虑使用多个线程池,这样每个线程池可以根据不同任务的工作负载进行调节.

计算密集型和I/O密集型的线程选择

计算密集型:一直在计算,cpu利用率高,过多的线程没有意义,反而切换线程会消耗额外的资源.

I/O密集型:例如查找数据库,等待数据造成的阻塞,CPU利用率低,多个线程可以提高响应速度.

对于计算密集型的任务,一个有N个处理器的系统通常使用一个N+1个线程的线程池来获得最优的利用率(计算密集型的线程恰好在某时因为发生一个页错误或者因其它原因而暂停,刚好有一个"额外"的线程,可以确保在这种情况下CPU周期不会中断工作).

对于包含了I/O和其他阻塞操作的任务,不是所有的线程都会在所有的时间被调度,因此你需要一个更大的池.

在一个基准负载下,可以使用不同大小的线程池运行你的应用程序,并观察CPU利用率的水平.

计算线程池大小的公式

N = CPU的数量

U = 目标CPU的使用率,介于0-1之间

W/C = 等待时间(wait)和计算时间(calculate)的比率

为保持处理器到达期望的使用率,最优的池的大小等于:

num(线程数) = N * U * (1 + W/C);

你可以使用Runtime来获得CPU的数目:

int nCpus = Runtime.getRuntime().availableProcessors();

简单的例子

通过Runtime.getRuntime().availableProcessors();得到我的电脑cpu数是4, 我期望cpu的使用率是100%,假设等待时间是10秒,计算时间是1秒.那么我最优的池大小就是:

4 * 100% * (1+10/1) = 44

线程池的长度和资源池的长度互相影响

当任务需要使用池化的资源时,比如数据库链接,线程池的长度和资源池的长度会互相影响.

如果每一个任务都需要一个数据库链接,那么连接池的大小就限制了线程池的有效大小;类似地,当线程池中的任务是连接池的唯一消费者时,那么线程池的大小反而又会限制了连接池的有效大小.

---

配置ThreadPoolExecutor

灵活配置ThreadPoolExecutor

使用Executors工厂方法可以创建各种类型的线程池,newCachedThreadPool、newFixedThreadPool和newScheduledThreadExecutor等.如果这些执行策略不能满足你的需求,你可以 new ThreadPoolExecutor(传递各种参数)来配置.

ThreadPoolExecutor有很多构造函数

最后一个构造函数是功能最全的,也是最常用的,源码:

 public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.acc = System.getSecurityManager() == null ?
                null :
                AccessController.getContext();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

它有五个参数分别是:

• corePoolSize 核心池大小
• maximumPoolSize 最大池大小
• keepAliveTime 存活时间
• TimeUnit 时间单元
• BlockingQueue 工作队列
• ThreadFactory 线程工厂
• RejectedExecutionHandler 拒绝执行处理器

注意源码,限定了设置这几个值的范围,不满足就会报非法参数异常,当时博主就是将核心池的值设置比最大池的值大,报了这个异常,看了源码才晓得:

 if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();

corePoolSize(核心池大小)、maximum pool size(最大池的大小)和存活时间(keep-alive time)共同管理着线程的创建与销毁.

corePoolSize: 线程池的实现试图维护池的大小(Eexcutors.newSingleThreadExecutor就是一种实现);即时没有执行任务,池的大小也等于核心池的大小,工作队列充满后会创建更多的线程.(Executors.newCacheThreadPool池的大小就是不固定的,随着任务增减线程的数量).

maximumPoolSize:最大池的大小制约着线程池可以同时执行的的最大线程数(限制并发数量),如果一个线程闲置的时间超过了存活时间就会被回收.并且同时运行的线程的数量不能超过核心池大小,线程池会终止超过的数量.

keep alive time & TimeUniht: 存活时间保证了空闲的线程会被回收,这样释放了这部分被占有的资源可以做更有用的事情.

两种特殊的线程池实现

Executors.newFixedThreadPool和newCachedThreadPool是两种特殊的实现.

newFixedThreadPool设置了核心池和最大池的大小,而且永远不会超时(不会回收线程).

newCachedThreadPool将最大池设置为了Integer.MAX_VALUE,核心池设置为0,超时设置为一分钟,这样创建出来的可无限扩大的线程池,会在需求量减少的情况下减少线程数量.

不要将核心池大小设置为0

我们自己手动创建ThreadPoolExecutor的时候,不要将corePoolSize设置为0,除非你用的是SynchronousQueue队列(newCachedThreadPool用的就是),否则队列不充满,就不会执行.

管理任务队列

用定长的线程池去替代为每个任务都创建一个线程的方式,在高负载的情况下,使得程序更不容易崩溃了,暂时没时间处理的任务会放进阻塞队列里,这是一个更优的方案,但是如果传递进来的任务超过处理的速速,程序仍然有可能崩溃.

过多的请求会使程序崩溃或者响应很慢

过多的请求会导致两个问题:

1. 耗尽内存
2. 响应速度很慢(后请求的用户需要等到前面的请求执行完)

队列可以缓和上述问题.

队列有助于缓和瞬时的任务激增,但是最终你还需要遏制请求的达到速率.

ThreadPoolExecutor的第五个参数是一个BlockingQueue阻塞队列.队列有三种:无限队列、有限队列和同步移交.队列的选择和很多其他的配置参数都有关系,比如池的大小.

newFixedThreadPool 和 newSingleThreadPool默认使用的是无界的LinkedBlockingQueue.任务无限多,队列无限长.最后就是程序崩溃.

所以我们最终的选择是使用有界队列,例如ArrayBlockingQueue或者有限的LinkedBlockingQueue以及PriorityBlockingQueue(自定义优先级的队列),有界队列满了以后有饱和策略可以处理那些没放进队列中的请求.

池的大小和队列的大小相辅相成

一个大队列加一个小池,可以控制对内存和CPU的使用,可以减少上下文切换,不过会影响吞吐量.

SynchronousQueue

对于庞大或者无限的池可以使用SynchronousQueue,这不是一个真正的队列,原来的队列可以理解为把任务存在一个容器里,线程放进去,线程取出来,而SynchronousQueue相当于是一个手递手传递.

两种情况可以使用SynchronousQueue:

• 任务可以被拒绝
• 池无限大

newCachedThreadPool使用的就是这个队列.

newCachedThreadPool和定长线程池之间的选择

如果不会因为负载过大导致程序崩溃就是用newCachedThreadPool,因为它的队列用的是SynchronousQueue,它有更佳的吞吐量.

还有一点要特别注意,如果任务之间相互依赖(一个任务依赖于另一个任务的结果,像这样的)最好使用newCachedThreadPool,否则可能产生线程饥饿死锁.

反之,像互联网应用程序还是应该使用定长的线程池.

饱和策略:处理未放入队列的任务

ThreadPoolExecutor的第七个参数,
可以用过调用setRejectedExecutionHandler来修改(如果线程池关闭了也会用到饱和策略).

Java提供了几种RejectedExecutionHandler实现:

• AbortPolicy
• CallerRunsPolicy
• DiscardPolicy
• DiscardOldestPolicy

默认的AbortPolicy会引起execute抛出未检查的RejectedExecutionException;调用者可以捕获这个异常,然后编写满足自己需求的处理代码(例如:持久化这个任务,一会再执行).

DiscardPolicy策略会默认放弃这个任务;

DiscardOldestPolicy会放弃本该接下来执行的任务,同时还会尝试去重新提交新任务.(无法和优先级队列同时使用,遗弃的任务是优先级最高的)

CallerRunsPolicy(调用者运行)既不会丢弃任务也不会抛出异常,它会把任务推送会调用者那里,以减缓新任务流,它不会再池线程中执行最新提交的任务,会在调用线程池execute或submit方法的线程中执行.

public class ThreadPool {

    private static class  Worker implements Runnable{
        @Override
        public void run() {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName()+" is running");
        }
    }


    public static void main(String [] args){
        //Executors.newSingleThreadExecutor()
        ExecutorService executorService = new ThreadPoolExecutor(2, 2, 1,
                TimeUnit.MICROSECONDS, new ArrayBlockingQueue<Runnable>(2), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r);
            }
        },new ThreadPoolExecutor.CallerRunsPolicy());


        for (int i = 0; i < 5; i++) {
            executorService.submit(new Worker());
        }

    }
}

打印输出:

main is running

Thread-1 is running

Thread-0 is running

Thread-0 is running

Thread-1 is running

证明在main方法的线程中执行了最新的任务.

执行100次代码中的循环会循环输出,main 和 Thread-1 Thread-0,相当于加上主线程三个线程并发执行.

当所有的池线程都被占用,而且工作队列已充满后,下一个任务会在主线程中执行.主线程执行任务的时候会花费一些时间,这时候主线程是不能提交任何任务的,所以这也给工作者线程一些时间来追赶进度.

这期间主线程不会调用accept接受新的请求,而会在TCP层的队列中等候.如果持续高负载的话,最终会由TCP层判断它的链接请求队列是否已经排满,如果已满就开始丢弃请求任务.

当服务器过载的时候,首先线程池里的所有线程都忙碌了起来,然后阻塞队列满了,接着TCP层满了,最终就是用户请求失败.

这使得服务器在高负载下可以平缓地劣化(graceful degradation).

线程工厂

ThreadPoolExecutor构造方法的第六个参数.

ThreadFactory接口只有一个方法 newThread,在线程池需要创建一个新线程时使用的.

定制这个工厂的用处:

• 为池线程指明一个UncaughtExceptionHandler(上篇博客有解释)
• 实例化一个定制Thread类的实例(例如给线程添加一个新的名称)
• 修改线程的优先级(不要这样做)
• 修改后台状态(不要这样做)

自定义线程池:

public class CustomThreadFactory implements ThreadFactory {

    private String poolName;

    public CustomThreadFactory(String poolName) {
        this.poolName = poolName;
    }

    @Override
    public Thread newThread(Runnable r) {
        return new MyThread(r,poolName);
    }
}

自定义线程:

public class MyThread extends Thread {

    private final Logger log = Logger.getAnonymousLogger();

    public MyThread(Runnable target,String name) {
        super(target,name);
        //线程异常终止的时候会得到记录.
        setUncaughtExceptionHandler(new UncaughtExceptionHandler() {
            @Override
            public void uncaughtException(Thread t, Throwable e) {
                log.info("Uncaught in thread"+t.getName()+e);
            }
        });
    }

    @Override
    public void run(){
        //可以做一些额外的日志记录
        super.run();

    }
}

重新设置线程池的参数

ThreadPoolExecutor的属性可以在创建后,通过setters方法重新设置.

如果线程池是通过Executors工厂方法创建的(除newSingleThreadExecutor以外),可以先转型为ThreadPoolExecutor,然后在设置.

public class UpdateExecutor {
    public static void main(String [] args){
        Executor executor = Executors.newCachedThreadPool();
        ((ThreadPoolExecutor)executor).setCorePoolSize(111);
        
    }
}

如果你不想你的线程池被修改可以使用Executors.unconfigurableExecutorService()方法.使线程池无法被修改

Executor executor = Executors.unconfigurableExecutorService(Executors.newCachedThreadPool());

扩展ThreadPoolExecutor

ThreadPoolExecutor提供了几个函数让子类去覆写来扩展ThreadPoolExecutor

• beforeExecute
• afterExecute
• terminate

beforeExecute在任务执行前调用,afterExecute在任务执行后调用,可以用它们来写日志.

无论任务是正常地从run返回,还是抛出一个异常,afterExecutor都会被调用(如果任务完成后抛出一个Error,afterExecute不会被调用).如果beforeExecutor抛出一个RuntimeException,任务不会被执行,afterExecutor也不会被调用.

terminated会在线程池关闭后调用.也就是当所有任务都已完成并且所有工作者线程都已经关闭后,会执行terminated.

示例:

public class CustomThreadPoolExecutor extends ThreadPoolExecutor {
    public CustomThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    public CustomThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
    }

    public CustomThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
    }

    public CustomThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        System.out.println("beforeExecute方法执行了");
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        System.out.println("afterExecute方法执行了");
    }


    @Override
    protected void terminated() {
        System.out.println("terminated方法执行了");
    }

    public static void main(String[] args) throws InterruptedException {
        CustomThreadPoolExecutor customThreadPoolExecutor = new CustomThreadPoolExecutor(1,
                1, 1, TimeUnit.SECONDS, new ArrayBlockingQueue<>(3));

                customThreadPoolExecutor.submit(()-> System.out.println("任务执行了"));

                customThreadPoolExecutor.shutdown();
                customThreadPoolExecutor.awaitTermination(1,TimeUnit.SECONDS);
    }

}

输出:
beforeExecute方法执行了
任务执行了
afterExecute方法执行了
terminated方法执行了

总结

对于并发执行的任务,Executor框架是强大且灵活的.它提供了大量可调节的选项,比如创建和关闭线程的策略,处理队列任务的策略,并且提供了几个钩子函数用于扩展它的行为.然而,和大多数的框架一样,草率地将一些设定组合在一起,并不能很好地工作;一些类型的任务需要特定的执行策略,而一些参数组合在一起后可能产生意外的后果.

下一篇会更新关于死锁的博客.

再见.