阻塞队列的应用之线程池

线程池基本概念

概念：线程池主要是控制运行线程的数量，将待处理任务放到等待队列，然后创建线程执行这些任务。如果超过了最大线程数，则等待。

优点：

  • 1.线程复用：不用一直new新线程，重复利用已经创建的线程来降低线程的创建和销毁开销，节省系统资源。

  • 2.提高响应速度：当任务达到时，不用创建新的线程，直接利用线程池的线程。

  • 3.管理线程：可以控制最大并发数，控制线程的创建等。

体系：

Executor→ExecutorService→AbstractExecutorService→ThreadPoolExecutor。ThreadPoolExecutor是线程池创建的核心类。类似Arrays、Collections工具类，Executor也有自己的工具类Executors。

线程池三种常用创建方式

newFixedThreadPool：使用LinkedBlockingQueue实现，定长线程池。

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

newSingleThreadExecutor：使用LinkedBlockingQueue实现，一池只有一个线程。

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

newCachedThreadPool：使用SynchronousQueue实现，变长线程池。

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

线程池创建的七个参数

  • 1.corePoolSize：线程池常驻核心线程数

  • 2.maximumPoolSize：能够容纳的最大线程数

  • 3.keepAliveTime：空闲线程存活时间

  • 4.unit：存活时间单位

  • 5.workQueue：存放提交但未执行任务的队列

  • 6.threadFactory：创建线程的工厂类

  • 7.handler：等待队列满后的拒绝策略

理解：线程池的创建参数，就像一个银行。

corePoolSize就像银行的“当值窗口“，比如今天有2位柜员在受理客户请求（任务）。如果超过2个客户，那么新的客户就会在等候区（等待队列workQueue）等待。当等候区也满了，这个时候就要开启“加班窗口”，让其它3位柜员来加班，此时达到最大窗口maximumPoolSize，为5个。如果开启了所有窗口，等候区依然满员，此时就应该启动”拒绝策略“handler，告诉不断涌入的客户，叫他们不要进入，已经爆满了。由于不再涌入新客户，办完事的客户增多，窗口开始空闲，这个时候就通过keepAlivetTime将多余的3个”加班窗口“取消，恢复到2个”当值窗口“。

线程池的底层原理

原理图：上面银行的例子，实际上就是线程池的工作原理。

流程图：

新任务到达→

如果正在运行的线程数小于corePoolSize，创建核心线程；大于等于corePoolSize，放入等待队列。

如果等待队列已满，但正在运行的线程数小于maximumPoolSize，创建非核心线程；大于等于maximumPoolSize，启动拒绝策略。

当一个线程无事可做一段时间keepAliveTime后，如果正在运行的线程数大于corePoolSize，则关闭非核心线程。

线程池的拒绝策略

当等待队列满时，且达到最大线程数，再有新任务到来，就需要启动拒绝策略。JDK提供了四种拒绝策略，分别是。

  • 1.AbortPolicy：默认的策略，直接抛出RejectedExecutionException异常，阻止系统正常运行。

  • 2.CallerRunsPolicy：既不会抛出异常，也不会终止任务，而是将任务返回给调用者。

  • 3.DiscardOldestPolicy：抛弃队列中等待最久的任务，然后把当前任务加入队列中尝试再次提交任务。

  • 4.DiscardPolicy：直接丢弃任务，不做任何处理。

实际生产使用哪一个线程池？

单一、可变、定长都不用！原因就是FixedThreadPool和SingleThreadExecutor底层都是用LinkedBlockingQueue实现的，这个队列最大长度为Integer.MAX_VALUE，显然会导致OOM。所以实际生产一般自己通过ThreadPoolExecutor的7个参数，自定义线程池。

ExecutorService threadPool=new ThreadPoolExecutor(2,5,
1L,TimeUnit.SECONDS,
new LinkedBlockingQueue<>(3),
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.AbortPolicy());

public class MyThreadPoolDemo {
public static void main(String[] args) {
System.out.println("Fixed Thread Pool");
fixedThreadPool();
System.out.println("Single Thread Pool");
singleThreadPool();
System.out.println("Cached Thread Pool");
cachedThreadPool();
System.out.println("Custom Thread Pool");
customThreadPool();
}
private static void customThreadPool() {
ExecutorService threadPool=
new ThreadPoolExecutor(2,
5,
1L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(3),
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.AbortPolicy()
);
try {
for (int i = 0; i < 9; i++) {
threadPool.execute(() -> {
System.out.println(Thread.currentThread().getName()+"t办理业务");
});
}
} catch (Exception e) {
e.printStackTrace();
} finally {
threadPool.shutdown();
}
}
private static void cachedThreadPool() {
//不定量线程
ExecutorService threadPool = Executors.newCachedThreadPool();
try {
for (int i = 0; i < 9; i++) {
threadPool.execute(() -> {
System.out.println(Thread.currentThread().getName()+"t办理业务");
});
}
} catch (Exception e) {
e.printStackTrace();
} finally {
threadPool.shutdown();
}
}
private static void singleThreadPool() {
//一池1个线程
ExecutorService threadPool = Executors.newSingleThreadExecutor();
try {
for (int i = 0; i < 9; i++) {
threadPool.execute(() -> {
System.out.println(Thread.currentThread().getName() + "t办理业务");
});
}
} catch (Exception e) {
e.printStackTrace();
} finally {
threadPool.shutdown();
}
}
private static void fixedThreadPool() {
//一池5个线程
ExecutorService threadPool = Executors.newFixedThreadPool(5);
//一般常用try-catch-finally
//模拟10个用户来办理业务，每个用户就是一个线程
try {
for (int i = 0; i < 9; i++) {
threadPool.execute(() -> {
System.out.println(Thread.currentThread().getName() + "t办理业务");
});
}
} catch (Exception e) {
e.printStackTrace();
} finally {
threadPool.shutdown();
}
}
}

自定义线程池参数选择

对于CPU密集型任务，最大线程数是CPU线程数+1。对于IO密集型任务，尽量多配点，可以是CPU线程数*2，或者CPU线程数/(1-阻塞系数)。

死锁编码和定位

public class DeadLockDemo {
//两个锁对象
private static Object obj1 = new Object();
private static Object obj2 = new Object();
public static void main(String[] args) {
new Thread(() -> {
while (true) {
synchronized (obj1) {
System.out.println(Thread.currentThread().getName() + "获得obj1锁");
try {
//睡3秒让另一个线程先获得obj2锁
TimeUnit.SECONDS.sleep(3);
//此时另一个线程已经持有obj2锁，这里请求获取obj2锁就会发生死锁
synchronized (obj2) {
System.out.println(Thread.currentThread().getName() + "获得obj2锁");
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}).start();
new Thread(() -> {
while (true) {
synchronized (obj2) {
System.out.println(Thread.currentThread().getName() + "获得obj2锁");
try {
//睡3秒让另一个线程先获得obj1锁
TimeUnit.SECONDS.sleep(3);
synchronized (obj1) {
System.out.println(Thread.currentThread().getName() + "获得obj1锁");
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}).start();
}
}

主要是两个命令配合起来使用，定位死锁。

jps指令：jps -l可以查看运行的Java进程。

jstack指令：jstack pid可以查看某个Java进程的堆栈信息，同时分析出死锁。

以上是 阻塞队列的应用之线程池 的全部内容， 来源链接： utcz.com/a/25400.html

回到顶部