Java多线程——线程池的使用

线程饥饿死锁

• 在单线程的Executor中，如果任务A将任务B提交给同一个Executor，并且等待任务B的结果，就会引发死锁
• 线程池中所有正在执行任务的线程由于等待其他仍处于工作队列中的任务而阻塞

执行时间较长的任务

执行时间较长的任务不仅会造成线程池阻塞，也会增加执行时间较短的任务的服务时间

对于平台库的可阻塞方法都定义了限时版本，如BlockingQueue.put，若超过等待时间，则可以中止任务或重新放回队列后续执行

设置线程池的大小

对于计算密集型的任务，在有N个处理器的系统上，线程池大小为N+1能实现最优利用率

对于其他情况

ThreadPoolExecutor

通过一系列参数配置线程池

• corePoolSize：核心线程数，当allowsCoreThreadTimeOut=true且核心线程闲置时间超过keepAliveTime就会终止，否则一直存活
• maximumPoolSize：最大线程数，超过后接下来的任务会被阻塞
• keepAliveTime：非核心线程闲置时间，超过后非核心线程会被回收，当allowsCoreThreadTimeOut=true同样作用于核心线程
• unit：指定keepAliveTime的时间单位
• workQueue：线程池的任务队列，存储execute()提交的Runnable对象
• threadFactory：线程工厂，为线程池创建新线程
• handler：当线程池已满或无法成功执行时，使用饱和策略

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.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;
}

管理队列任务

ThreadPoolExecutor允许提供一个BlockingQueue保持执行的任务，基本队列有3种

• 无界队列，newFixedThreadPool和newSingleThreadPool使用LinkedBlockingQueue
• 有界队列，如ArrayBlockingQueue，有界的LinkedBlockingQueue和PriorityBlockingQueue
• 同步移交，newCachedThreadPool使用SynchronousQueue，避免任务排队，直接将任务从生产者移交给工作者线程

只有当任务互相独立，线程池和工作队列设置界限才是合理的，否则有界的线程池或队列可能导致线程饥饿死锁问题

饱和策略

当有界队列被填满后，将会使用饱和策略RejectedExecutionHandler，其包含

• AbortPolicy：默认策略，抛出RejectedExecutionException，调用者需捕获异常并自行处理
• CallerRunsPolicy：将某些任务回退给调用者，在调用了execute的线程中执行该任务（如主线程提交任务会退回给主线程执行）
• DiscardPolicy：抛弃新提交的任务
• DiscardOlderstPolicy：抛弃下一个将被执行的任务，尝试重新提交新的任务

线程工厂

线程池通过线程工厂创建线程，默认线程工厂创建一个新的非守护线程，可通过重新ThreadFatory的newThread方法自定义配置信息，每次创建线程都会调用该方法

如下为线程池中的线程设置名字和setUncaughtExceptionHandler

class MyThreadFactory implements ThreadFactory {@Overridepublic Thread newThread(Runnable r) {Thread thread = new Thread("AAA");thread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {@Overridepublic void uncaughtException(@NonNull Thread t, @NonNull Throwable e) {System.out.println(e.toString());}});return thread;}
}

修改线程池配置

可将对象转为ThreadPoolExecutor，再调用对应属性的setter方法，但newSingleThreadExecutor除外，其使用unconfigurableExecutorService封装，只暴露了ExecutorService中的方法

public class A {public ExecutorService test() {ExecutorService cachedThreadPool = Executors.newCachedThreadPool();if (cachedThreadPool instanceof ThreadPoolExecutor) {((ThreadPoolExecutor) cachedThreadPool).setCorePoolSize(10);}return Executors.unconfigurableExecutorService(cachedThreadPool);}
}

扩展ThreadPoolExecutor

可在子类中重写

• beforeExecute，线程执行前调用
• afterExecute，线程执行完后调用，无论线程正常返回还是抛出异常都会调用（但任务完成后出现Error则不会）
• terminated，线程池关闭时调用