您现在的位置是：首页 > 后端

当前栏目

Java线程池原理与使用详解

JAVA 线程原理详解使用

2023-09-11 14:21:24 时间

说明

在Java中，创建和销毁对象是很费时间的，因为创建一个对象要获取内存资源或者其它更多资源，且虚拟机将试图跟踪每一个对象，以便能够在对象销毁后进行垃圾回收。所以如果并发的线程数量很多，并且每个线程都是执行一个时间很短的任务就结束了，这样频繁创建销毁会大大降低系统性能。线程池的目的就是将线程复用，统一管理，以减少这类消耗，从而提高性能。

内容

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;
    }

创建一个线程池需要输入几个参数：

1、corePoolSize：核心线程的数量。在创建了线程池后，默认情况下，线程池中并没有任何线程，而是等待有任务到来才创建线程去执行任务，除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法来预创建线程，在没有任务到来之前就创建corePoolSize个线程或者一个线程。

2、maximumPoolSize：线程池中最大线程数量，线程池允许创建的最大线程数。这个参数大部分时候都与corePoolSize数值相同，具体还得是业务场景需求而定。

3、keepAliveTime：表示线程空闲多久时间会终止。默认情况下，只有当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用，直到线程池中的线程数不大于corePoolSize，即默认只作用于池中的非核心线程。但是如果调用了allowCoreThreadTimeOut(boolean)方法，在线程池中的线程数不大于corePoolSize时，keepAliveTime参数也会起作用，直到线程池中的线程数为0。

4、TimeUnit：为参数keepAliveTime的时间单位。

5、 workQueue：用于保存等待执行的任务的阻塞队列。可以选择以下几个阻塞队列。

5.1、ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按 FIFO（先进先出）原则对元素进行排序。

5.2、LinkedBlockingQueue：一个基于链表结构的阻塞队列，此队列按FIFO （先进先出）排序元素，吞吐量通常要高于ArrayBlockingQueue。

5.3、SynchronousQueue：一个不存储元素的阻塞队列。因为没有存储功能，因此put和take会一直阻塞，直到有另一个线程已经准备好参与到交付过程中。仅当有足够多的消费者，并且总是有一个消费者准备好获取交付的工作时，才适合使用同步队列。

5.4、PriorityBlockingQueue：一个具有优先级得无限阻塞队列，在构造函数需传入comparator,用于插入元素时继续排序，若没有传入comparator，则插入的元素必须实现Comparatable接口。

6、ThreadFactory：线程工厂，主要用来创建线程。

7、RejectedExecutionHandler：饱和策略，当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。当然也可以根据应用场景需要来实现RejectedExecutionHandler接口自定义策略。如记录日志或持久化不能处理的任务。

7.1、AbortPolicy：表示无法处理新任务时抛出异常，默认策略。

7.2、CallerRunsPolicy：直接在 execute 方法的调用线程中运行被拒绝的任务。

7.3、DiscardOldestPolicy：丢弃队列最前面的任务，然后尝试执行当前任务。

7.4、DiscardPolicy：不处理直接丢弃掉。

代码开发原理

public class MyThread implements Runnable{
 
        private String str1;
        public MyThread(String str){
            str1 = str;
        }
        @Override
        public void run() {
            System.out.println(str1);
        }
    }
 
//创建自带java线程池
        ExecutorService exe = Executors.newFixedThreadPool(15);
        for(int i=0;i<=5;i++){
            exe.submit(new MyThread("你好"+i)); //execute也可以
        }

Executors可以创建四种线程池

newCachedThreadPool,
newFixedThreadPool,
newScheduledThreadPool,
newSingleThreadExecutor

//自己创建线程池
public Executor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
		//核心线程数
        executor.setCorePoolSize(20);
        executor.setMaxPoolSize(50);
        executor.setQueueCapacity(100);
        executor.setKeepAliveSeconds(30);
        //让核心子线程超过30秒自动销毁
        executor.setAllowCoreThreadTimeOut(true);
        //对拒绝task的处理策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
	 Executor exe = taskExecutor();
        for(int i=0;i<=5;i++){
            exe.execute(new MyThread("你好"+i));
        }

向线程池提交任务及源码分析

1、使用execute提交任务，这个方法是ThreadPoolExecutor的核心方法，通过这个方法可以向线程池提交一个任务，交由线程池去执行，但是execute方法没有返回值，所以无法判断任务知否被线程池执行成功。

2、使用submit提交任务，这个方法也是用来向线程池提交任务的，但是它和execute()方法不同，它能够返回任务执行的结果，去看submit()方法的实现，会发现它实际上还是调用的execute()方法，只不过它利用了Future来获取任务执行结果。

具体看下execute：

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            reject(command);
    }

这里说明下两个点：

Worker和Task的概念：两者都是runnable，Worker是当前线程池中的线程，可以看做为工作人员，而Task虽然是runnable，但是并没有真正执行，而只是被Worker调用了run方法，相当于工作人员的任务。
maximumPoolSize，corePoolSize和workQueue的概念：corePoolSize是核心线程池的大小，决定了线程池里的“常驻工作人员”Worker的数量；workQueue是任务队列，相当于“工作人员”Worker的任务量；maximumPoolSize是线程池最大容量，就是池里能容纳“工作人员”Worker的最大数量，上面说了往往它和 corePoolSize一样大，但如果在一段并发高峰的时段，这时maximumPoolSize设置比corePoolSize大的作用就有了体现，在所有核心线程满负荷工作，及workQueue任务堆积满了的情况下，线程池可以创建一定数量（maximumPoolSize-corePoolSize）“临时工”Woker去处理溢出的任务，等过了高峰时段，“临时工”Woker过了空闲时间就会被回收，从而不会过多占用系统资源。

线程池执行流程

来看execute核心方法：addWorker
Worker的增加和Task的获取以及终止都是在此方法中实现的，也就是这一个方法里面包含了很多东西。在addWorker方法中提到了runState的概念，runState是线程池的核心概念，我们先看下相关声明及方法：

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;
 
    // runState is stored in the high-order bits
    private static final int RUNNING    = -1 << COUNT_BITS;
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    private static final int STOP       =  1 << COUNT_BITS;
    private static final int TIDYING    =  2 << COUNT_BITS;
    private static final int TERMINATED =  3 << COUNT_BITS;
 
    // Packing and unpacking ctl
    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包含了：runState（线程池当前状态）和workerCount（工作线程worker的数量），其中ctl的前三位代表runState，后29位代表workerCount。通过runStateOf和workerCountOf能分别获取线程池状态和工作线程worker的数量。

runState代表整个线程池的运行生命周期，有如下取值：
1.RUNNING：可以新加任务，同时可以处理queue中的任务。
2.SHUTDOWN：不增加新任务，但是处理queue中的任务。
3.STOP：不增加新任务，同时不处理queue中的任务。
4.TIDYING：所有的任务都终止处理了（queue中），同时workerCount为0，那么此时进入TIDYING
5.terminated()：方法结束，变为TERMINATED

继续看addWorker的代码：

private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
 
            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
 
            for (;;) {
                int wc = workerCountOf(c);
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }
 
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int rs = runStateOf(ctl.get());
 
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

按流程下来，先是判断runState是否为running，不是的话在判断是否为shutdown，若是则firstTask与workQueue是否为空，这一系列判断对应了上面提到的线程池处于什么状态所执行任务的逻辑。

接着就是判断没有到corePoolSize 或maximumPoolSize上限时，那么允许添加工作线程，CAS增加Worker的数量后，跳出循环。

后面就是实例化Worker，并且将workers是HashSet线程不安全的，那么此时需要加锁，所以mainLock.lock(); 之后重新检查线程池的状态，如果状态不正确，那么调用addWorkerFailed减小Worker的数量。如果状态正常，那么添加Worker到workers。添加成功后即调用Worker中Thread的start方法，以执行Worker中的runWorker方法。

接下来看runWorker方法的代码：

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            while (task != null || (task = getTask()) != null) {
                w.lock();
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                        task.run();
                    } catch (RuntimeException x) {
                        thrown = x; throw x;
                    } catch (Error x) {
                        thrown = x; throw x;
                    } catch (Throwable x) {
                        thrown = x; throw new Error(x);
                    } finally {
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }

这个方法就是让工作线程Worker处理任务，任务来源有两处：一是firstTask，就是新建Worker时提交的优先任务；二是通过getTask方法去workQueue获取任务。

这里说下beforeExecute与afterExecute，这两个方法适用于拓展的，通过自己继承ThreadPoolExecutor实现线程池类，重写这两个方法，可以作日志打印，运行情况统计等功能。

最后就是重点的getTask：

private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?
 
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }
            int wc = workerCountOf(c);
            // Are workers subject to culling?
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
            try {
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

这里开始也是照常判断runState的状态，再判断Worker是否需要过期，true的话使用poll取任务。如果正常返回，那么返回取到的task。如果超时，证明Worker空闲，同时Worker超过了corePoolSize，需要删除。false的话则是take取任务，有的话返回，没有获取到时将一直阻塞，知道获取到或者中断。

线程池的关闭

有两个方法可以提供线程池的关闭，分别是shutDown 和 shutDownNow。但是它们的实现原理不同，shutdown的原理是只是将线程池的状态设置成SHUTDOWN状态，然后中断所有没有正在执行任务的线程。shutdownNow的原理是遍历线程池中的工作线程，然后逐个调用线程的interrupt方法来中断线程，所以无法响应中断的任务可能永远无法终止。shutdownNow会首先将线程池的状态设置成STOP，然后尝试停止所有的正在执行或暂停任务的线程，并返回等待执行任务的列表。

如何构建健壮的线程池

如何合理的估算线程池大小

在创建线程池时，几乎应该都碰见过这问题，设置核心线程池多大合适，阻塞队列设置多大合适。
《Java并发编程实战》中有提到：如果是CPU密集型应用，则线程池大小设置为N+1；如果是IO密集型应用，则线程池大小设置为2N+1；N为CPU数目。当然这只是初略的估算。书中还提到了更为详细的计算公式：

N [CPU数目]P[CPU利用率]（1+I[等待时间] / C[计算时间]）
其中等待时间I就是线程中发生IO等阻塞操作是花费的时间，计算时间C则是线程执行计算花费的时间，可以看出，等待时间越长，为了充分利用cpu，线程池大小就应该越大。

辅助计算线程池大小程序

在网上找到的一个估算方法PoolSizeCalculator ，先看代码（来源：https://github.com/nschlimm/playground/blob/master/java7-playground/src/main/java/com/schlimm/java7/nio/threadpools/PoolSizeCalculator.java）：

/** * Calculates the boundaries of a thread pool for a given {@link Runnable}. * * @param targetUtilization * the desired utilization of the CPUs (0 <= targetUtilization <= 1) * @param targetQueueSizeBytes * the desired maximum work queue size of the thread pool (bytes) */
	protected void calculateBoundaries(BigDecimal targetUtilization, BigDecimal targetQueueSizeBytes) {
		calculateOptimalCapacity(targetQueueSizeBytes);
		Runnable task = creatTask();
		start(task);
		start(task); // warm up phase
		long cputime = getCurrentThreadCPUTime();
		start(task); // test intervall
		cputime = getCurrentThreadCPUTime() - cputime;
		long waittime = (testtime * 1000000) - cputime;
		calculateOptimalThreadCount(cputime, waittime, targetUtilization);
	}

这里方法的两个入参，分别是期望的CPU利用率（0-1），及期望的阻塞任务队列的内存大小（byte）。calculateOptimalCapacity是计算预期队列的任务数量。

private void calculateOptimalCapacity(BigDecimal targetQueueSizeBytes) {
	long mem = calculateMemoryUsage();
	BigDecimal queueCapacity = targetQueueSizeBytes.divide(new BigDecimal(mem), RoundingMode.HALF_UP);
	System.out.println("Target queue memory usage (bytes): " + targetQueueSizeBytes);
	System.out.println("createTask() produced " + creatTask().getClass().getName() + " which took " + mem
			+ " bytes in a queue");
	System.out.println("Formula: " + targetQueueSizeBytes + " / " + mem);
	System.out.println("* Recommended queue capacity (bytes): " + queueCapacity);
}
	
public long calculateMemoryUsage() {
	BlockingQueue<Runnable> queue = createWorkQueue();
	for (int i = 0; i < SAMPLE_QUEUE_SIZE; i++) {
		queue.add(creatTask());
	}
	long mem0 = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();
	long mem1 = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();
	queue = null;
	collectGarbage(15);
	mem0 = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();
	queue = createWorkQueue();
	for (int i = 0; i < SAMPLE_QUEUE_SIZE; i++) {
		queue.add(creatTask());
	}
	collectGarbage(15);
	mem1 = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();
	return (mem1 - mem0) / SAMPLE_QUEUE_SIZE;
}

start与getCurrentThreadCPUTime是为了计算线程运行的等待时间与计算时间，后者可以通过ManagementFactory.getThreadMXBean().getCurrentThreadCpuTime()实现

public void start(Runnable task) {
		long start = 0;
		int runs = 0;
		do {
			if (++runs > 5) {
				throw new IllegalStateException("Test not accurate");
			}
			expired = false;
			start = System.currentTimeMillis();
			Timer timer = new Timer();
			timer.schedule(new TimerTask() {
				public void run() {
					expired = true;
				}
			}, testtime);
			while (!expired) {
				task.run();
			}
			start = System.currentTimeMillis() - start;
			timer.cancel();
		} while (Math.abs(start - testtime) > EPSYLON);
		collectGarbage(3);
	}

最后说下使用及效果，这里cpu利用率设置的是0.5，队列大小为100000byte：

public class SimplePoolSizeCaculatorImpl extends PoolSizeCalculator {
 
    @Override
    protected Runnable creatTask() {
        return new AsyncIOTask();
    }
 
    @Override
    protected BlockingQueue createWorkQueue() {
        return new LinkedBlockingQueue(1000);
    }
 
    @Override
    protected long getCurrentThreadCPUTime() {
        return ManagementFactory.getThreadMXBean().getCurrentThreadCpuTime();
    }
 
    public static void main(String[] args) {
        PoolSizeCalculator poolSizeCalculator = new SimplePoolSizeCaculatorImpl();
        poolSizeCalculator.calculateBoundaries(new BigDecimal(0.5), new BigDecimal(100000));
    }
 
}

任务样例：

class AsyncIOTask implements Runnable {
    @Override
    public void run() {
        HttpURLConnection connection = null;
        BufferedReader reader = null;
        try {
            String getURL = "http://baidu.com";
            URL getUrl = new URL(getURL);
            connection = (HttpURLConnection) getUrl.openConnection();
            connection.connect();
            reader = new BufferedReader(new InputStreamReader(
                    connection.getInputStream()));
            String line;
            while ((line = reader.readLine()) != null) {
// System.out.println(line);
            }
        }
        catch (IOException e) {
        } finally {
            if(reader != null) {
                try {
                    reader.close();
                }
                catch(Exception e) {
                }
            }
            connection.disconnect();
        }
    }
}

运行结果，上面号开头的是计算出来的队列大小，下面号计算出来的是线程池大小：

注：如果一直运行失败，可尝试把PoolSizeCalculator 中的精准度EPSYLON属性设置大点，如50。

参考资料

深入理解java线程池—ThreadPoolExecutor
https://www.jianshu.com/p/ade771d2c9c0
如何合理地估算线程池大小
http://ifeve.com/how-to-calculate-threadpool-size/
Threading stories: about robust thread pools
http://niklasschlimm.blogspot.com/2012/03/threading-stories-about-robust-thread.html#more
Determining Memory Usage in Java
https://www.javaspecialists.eu/archive/Issue029.html

猜你喜欢

mac上使用polipo将socks5代理转为http代理
记录一次保持原电压实现内存超频
js 判断数组中是否包含某个元素（转载）
Unity Loom多线程
js变量常量作用域
ubuntu12.04更新软件源时出现校验和不符
什么是 Native、Web App、Hybrid、React Native 和 Weex？（转载）
用上大数据，一级市场的投资人也可以通过量化找项目了？
Mozilla安全开源基金：让Heartbleed历史不再重演
Java集合转有类型的数组之toArray(T[] a)
PostgreSQL 异步IO实测
2021年春季学期-信号与系统-第四次作业参考答案-第三小题
【Web漏洞探索】文件包含漏洞
sharding-jdbc学习
CSRF 攻击的应对之道
【嵌入式Linux学习】6、交叉编译环境的搭建，单文件编译Hello Linux！
QtCreator 可以通过 Clang-Tidy 和 CLazy 对你的代码进行静态检查
充分发挥视频会议功效需学习合理选择设备
C# string reference type?
超融合系统市场规模增长79%，将步入主流应用
tkinter里面按钮如何按一次就失效
外网主机如何将数据包发送到共用一个公网IP的局域网某特定主机上的
排序算法：冒泡排序、插入排序、选择排序、希尔排序
爬虫之urllib库的使用

相关主题

java基本类型
Java 方法method
java线程基础
Java线程与锁-5
Java之线程池
Java中的线程
java线程池原理
java线程用法
java创建线程
java啊java
java：Calendar类

zl程序教程