java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > Java 线程池

超详细讲解Java线程池

作者:拉夫德鲁

本文主要介绍了Java线程池,本文运用大量代码和图片讲解相关知识,感兴趣的小伙伴一起来看看吧

带着问题阅读

1、什么是池化,池化能带来什么好处

2、如何设计一个资源池

3、Java的线程池如何使用,Java提供了哪些内置线程池

4、线程池使用有哪些注意事项

池化技术

池化思想介绍

池化思想是将重量级资源预先准备好,在使用时可重复使用这些预先准备好的资源。

池化思想的核心概念有:

例如大街上的共享单车,用户扫码开锁,使用完后归还到停放点,下一个用户可以继续使用,共享单车由厂商统一管理,为用户节省了购买单车的开销。

池化技术的应用

常见的池化技术应用有:资源池、连接池、线程池等。

在各种电商平台大促活动时,平台需要支撑平时几十倍的流量,因此各大平台在需要提前准备大量服务器进行扩容,在活动完毕以后,扩容的服务器资源又白白浪费。将计算资源池化,在业务高峰前进行分配,高峰结束后提供给其他业务或用户使用,即可节省大量消耗,资源池化也是云计算的核心技术之一。

网络连接的建立和释放也是一个开销较大的过程,提前在服务器之间建立好连接,在需要使用的时候从连接池中获取,使用完毕后归还连接池,以供其他请求使用,以此可节省掉大量的网络连接时间,如数据库连接池、HttpClient连接池。

线程的建立销毁都涉及到内核态切换,提前创建若干数量的线程提供给客户端复用,可节约大量的CPU消耗以便处理业务逻辑。线程池也是接下来重点要讲的内容。

如何设计一个线程池

设计一个线程池,至少需要提供的核心能力有:

基于以上角度,我们来分析Java是如何设计线程池功能的。

Java线程池解析

ThreadPoolExecutor使用介绍

大象装冰箱总共分几步

// 1.创建线程池
ThreadPoolExecutor threadPool = 
    new ThreadPoolExecutor(1, 1, 1L, TimeUnit.MINUTES, new LinkedBlockingQueue<>());
// 2.提交任务
threadPool.execute(new Runnable() {
    @Override
    public void run() {
        System.out.println("task running");
    }
}});
// 3.关闭线程池
threadPool.shutDown();

Java通过ThreadPoolExecutor提供线程池的实现,如示例代码,初始化一个容量为1的线程池、然后提交任务、最后关闭线程池。

ThreadPoolExecutor的核心方法主要有

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

类型 作用
ArrayBlockingQueue 数组结构的有界阻塞队列
LinkedBlockingQueue 链表结构的阻塞队列,可设定是否有界
SynchronousQueue 不存储元素的阻塞队列,直接将任务提交给线程池执行
PriorityBlockingQueue 支持优先级的无界阻塞队列
DelayQueue 支持延时执行的无界阻塞队列

类型 作用
AbortPolicy 拒绝并抛出异常。默认
CallerRunsPolicy 由提交任务的线程执行任务
DiscardOldestPolicy 抛弃队列头部任务
DiscardPolicy 抛弃该任务

// 提交Runnable
void execute(Runnable command);

// 提交Callable并返回Future
<T> Future<T> submit(Callable<T> task);

// 提交Runnable,执行结束后Future.get会返回result
<T> Future<T> submit(Runnable task, T result);

// 提交Runnable,执行结束后Future.get会返回null
Future<?> submit(Runnable task);
// 不再接收新任务,等待剩余任务执行完毕后停止线程池
void shutdown();

// 不再接收新任务,并尝试中断执行中的任务,返回还在等待队列中的任务列表
List<Runnable> shutdownNow();

内置线程池使用

To be useful across a wide range of contexts, this class provieds many adjustable parameters and extensibility hooks. However, programmers are urged to use the more convenient {@link Executors} factory methods {@link Executors#newCachedThreadPool} (unbounded thread poll, with automatic thread reclamation), {@link Executors#newFixedThreadPool} (fixed size thread pool) and {@link Executors#newSingleThreadExecutor}(single background thread), that preconfigure settings for the most common usage scenarios.

由于ThreadPoolExecutor参数复杂,Java提供了三种内置线程池newCachedThreadPoolnewFixedThreadPoolnewSingleThreadExecutor应对大多数场景。

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, 
                                  new SynchronousQueue<Runnable>());
}
public static ExecutorService new FixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads nThreads, 0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newSingleThreadExecutor() {
    return new FinallizableDelegatedExecutorService(
      (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, 
                                new LinkedBlockQueue<Runnable>()))
    );
}

ThreadPoolExecutor解析

整体设计

ThreadPoolExecutor基于ExecutorService接口实现提交任务,未采取常规资源池获取/归还资源的形式,整个线程池和线程的生命周期都由ThreadPoolExecutor进行管理,线程对象不对外暴露;ThreadPoolExecutor的任务管理机制类似于生产者消费者模型,其内部维护一个任务队列和消费者,一般情况下,任务被提交到队列中,消费线程从队列中拉取任务并将其执行。

线程池生命周期

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

private static int runStateOf(int c)     { return c & ~CAPACITY; } //计算当前运行状态
private static int workerCountOf(int c)  { return c & CAPACITY; }  //计算当前线程数量
private static int ctlOf(int rs, int wc) { return rs | wc; }   //通过状态和线程数生成ctl

TreadPoolExecutor通过ctl维护线程池的状态和线程数量,其中高3位存储运行状态,低29位存储线程数量。

线程池设定了RUNNINGSHUTDOWNSTOPTIDYINGTERMINATED五种状态,其转移图如下:

在这5种状态中,只有RUNNING时线程池可接收新任务,其余4种状态在调用shutDownshutDownNow后触发转换,且在这4种状态时,线程池均不再接收新任务。

任务管理解析

// 用于存放提交任务的队列
private final BlockingQueue<Runnable> workQueue;

// 用于保存池内的工作线程,Java将Thread包装成Worker存储
private final HashSet<Worder> workers = new HashSet<Worker>();

ThreadPoolExecutor主要通过workQueueworkers两个字段用于管理和执行任务。

线程池任务执行流程如图,结合ThreadPoolExecutor.execute源码,对任务执行流程进行说明:

if (workerCountOf(c) < corePoolSize) {
    // 创建新worker执行任务,true表示核心线程
    if (addWorker(command, true))
        return;
    c = ctl.get();
}
// 放入workQueue队列
if (isRunning(c) && workQueue.offer(command)) {
    int recheck = ctl.get();
    // 这里采用double check再次检测线程池状态
    if (! isRunning(recheck) && remove(command))
        reject(command);
    // 避免加入队列后,所有worker都已被回收无可用线程
    else if (workerCountOf(recheck) == 0)
        addWorker(null, false);
}
// 接上,放入队列失败
// 添加新worker执行任务,false表示非核心线程
else if (!addWorker(command, false))
    // 如添加失败,执行拒绝策略
    reject(command);

woker对象

ThreadPoolExecutor没有直接使用Thread记录线程,而是定义了worker用于包装线程对象。

private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
    ...
    final Thread thread;
    
    Runnable firstTask;
    
    Worker(Runnable firstTask) {
        setState(-1); // inhibit interrupts until runWorker
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }
    
    // worker对象被创建后就会执行
    public void run() {
        runWorker(this);
    }
}

worker对象通过addWorker方法创建,一般会为其指定一个初始任务firstTask,当worker执行完毕以后,worker会从阻塞队列中读取任务,如果没有任务,则该worker会陷入阻塞状态给出worker的核心逻辑代码:

private boolean addWorker(Runnable firstTask, boolean core) {
    ...
    // 指定firstTask,可能为null
    w = new Worker(firstTask);
    ...
    if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) {
        if (t.isAlive()) // precheck that t is startable
            throw new IllegalThreadStateException();
        workers.add(w);
        workerAdded = true;
    }
    ...
    // 执行新添加的worker
    if (workerAdded) {
        t.start();
        workerStarted = true;
    }
}


final void runWorker(Worker w) {
    // 等待workQueue的任务
    while (task != null || (task = getTask()) != null) {
    	...
    }
}

private Runnable getTask() {
    ...
    for (;;) {
        ...
        // 如果是普通工作线程,则根据线程存活时间读取阻塞队列
        // 如果是核心工作线程,则直接陷入阻塞状态,等待workQueue获取任务
        Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
        ...
    }
}

如下图,任务提交后触发addWorker创建worker对象,该对象执行任务完毕后,则循环获取队列中任务等待执行。

Java线程池实践建议

不建议使用Exectuors

线程池不允许使用Executors去创建,而是通过ThreadPoolExecutor的方式,这样的处理方式让写的同学更加明确线程池的运行规则,规避资源耗尽的风险。《阿里巴巴开发手册》

虽然Java推荐开发者直接使用Executors提供的线程池,但实际开发中通常不使用。主要考虑问题有:

CachedThreadPool将最大数量设置为Integer.MAX_VALUE,如果一直提交任务,可能造成Thread对象过多引起OOMFixedThreadPoolSingleThreadPoo的队列LinkedBlockingQueue无容量限制,阻塞任务过多也可能造成OOM

由于未指定ThreadFactory,线程名称默认为pool-poolNumber-thread-thredNumber,线程出现问题后不便定位具体线程池。

通常在完善的项目中,由于线程是重量资源,因此线程池由统一模块管理,重复创建线程池容易造成资源分散,难以管理。

线程池大小设置

通常按照IO繁忙型和CPU繁忙型任务分别采用以下两个普遍公式。

在理论场景中,如一个任务IO耗时40ms,CPU耗时10ms,那么在IO处理期间,CPU是空闲的,此时还可以处理4个任务(40/10),因此理论上可以按照IO和CPU的时间消耗比设定线程池大小。

《JAVA并发编程实践》中还考虑数量乘以目标CPU的利用率

在实际场景中,我们通常无法准确测算IO和CPU的耗时占比,并且随着流量变化,任务的耗时占比也不能固定。因此可根据业务需求,开设线程池运维接口,根据线上指标动态调整线程池参数。

推荐参考第二篇美团线程池应用

线程池监控

ThreadPoolExecutor提供以下方法监控线程池:

到此这篇关于超详细讲解Java线程池的文章就介绍到这了,更多相关Java 线程池内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
阅读全文