参考于:https://www.bilibili.com/video/BV1py4y1m7N6?from=search&seid=18247579430377169458&spm_id_from=333.337.0.0

快速导航

对比单线程、多线程的执行效率

测试3种线程池执行效率 以及原因

为什么阿里巴巴开发手册抵触使用Java自带线程池

线程的提交优先级、执行优先级

猜测一下,打印的“线程测试”的两个方法 那个是多线程执行的?

/**
 * @author : zanglikun
 * @date : 2021/12/31 15:25
 * @Version: 1.0
 * @Desc : 继承Thread 重写run方法
 */
public class ThreadDemo extends Thread{

    @Override
    public void run() {
        System.out.println("线程测试");
    }

    public static void main(String[] args) {
        new ThreadDemo().run();
        new ThreadDemo().start();
    }
}

答案是 start()执行的是多线程的。

我们Debug看一下:

run()方法

进入断点之后点击 照相机(获取线程转储) 就可以看到当前项目的线程信息了!

Start()方法

对比一下,run方法的线程名叫mainstart方法执行的线程名是 Thread-0@509

在对比一下 单线程与多线程的执行效率

单线程代码

    @SneakyThrows
    @Test
    public void testSingleThread() {
        long StartTime = System.currentTimeMillis();
        final ArrayList<String> arrayList = new ArrayList();
        for (int i = 0; i < 100000; i++) {
            Thread thread = new Thread() {
                @Override
                public void run() {
                    arrayList.add("abcdfg");
                }
            };
            thread.start();
            // join()让Thread这个线程 在main线程结束时等待,让其在main线程结束后在结束!
            thread.join();
        }
        long EndTime = System.currentTimeMillis();
        System.out.println("消耗了:" + (EndTime - StartTime) / 1000.00 + "秒");
        System.out.println(arrayList.size());
    }
消耗了:25.095秒
100000

多线程代码

    @SneakyThrows
    @Test
    public void testMuiltThread() {
        long StartTime = System.currentTimeMillis();
        final ArrayList<String> arrayList = new ArrayList();
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 100000; i++) {
            executorService.execute(() -> {
                arrayList.add("abcdfg");
            });
        }
        // 如果不shutdown,那么main方法走完了,直接打印消耗时间、打印集合长度时不准确的。因为main线程结束,线程池的线程还在执行,就打印结果肯定是不准确的
        executorService.shutdown();
        executorService.awaitTermination(1, TimeUnit.DAYS);
        long EndTime = System.currentTimeMillis();
        System.out.println("消耗了:" + (EndTime - StartTime) / 1000.00 + "秒");
        System.out.println(arrayList.size());
    }
消耗了:0.05秒
100000

测试不同线程池的效率

    @SneakyThrows
    @Test
    public void testDifferentThreadPool() {
        long StartTime = System.currentTimeMillis();
        final ArrayList<String> arrayList = new ArrayList();
        ExecutorService executorService1 = Executors.newCachedThreadPool(); //最快
        ExecutorService executorService2 = Executors.newFixedThreadPool(10); //慢大约49倍的执行时间(只在本场景(循环1000次)适用,不是两个线程池的的规范的效率差距)
        ExecutorService executorService3 = Executors.newSingleThreadExecutor(); //对比上一个慢了10倍(只在本场景(循环1000次)适用,不是两个线程池的的规范的效率差距)

        for (int i = 0; i < 1000; i++) {
            executorService1.execute(() -> {
                try {
                    System.out.println("当前线程名称"+Thread.currentThread().getName());
                    arrayList.add("abcdef");
                    Thread.sleep(500L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        // 如果不shutdown,那么main方法走完了,直接打印消耗时间、打印集合长度时不准确的。因为main线程结束,线程池的线程还在执行,就打印结果肯定是不准确的
        executorService1.shutdown();
        executorService1.awaitTermination(1, TimeUnit.DAYS);
        long EndTime = System.currentTimeMillis();
        System.out.println("消耗了:" + (EndTime - StartTime) / 1000.00 + "秒");
        System.out.println(arrayList.size());
    }

使用不同线程池分别执行一下,结果:

newCachedThreadPool() 0.616秒
打印日志如下:
当前线程名称pool-1-thread-1
当前线程名称pool-1-thread-3
当前线程名称pool-1-thread-2
......
当前线程名称pool-1-thread-997
当前线程名称pool-1-thread-998
当前线程名称pool-1-thread-999
当前线程名称pool-1-thread-1000
... 线程ID 编号取值1-1000



newFixedThreadPool() 50.037秒
打印日志如下:
当前线程名称pool-2-thread-1
当前线程名称pool-2-thread-2
当前线程名称pool-2-thread-3
当前线程名称pool-2-thread-4
当前线程名称pool-2-thread-6
当前线程名称pool-2-thread-5
当前线程名称pool-2-thread-7
当前线程名称pool-2-thread-8
当前线程名称pool-2-thread-10
当前线程名称pool-2-thread-9
... 以此循环线程ID 编号取值1-10  这里就能看到线程复用的提现!




newSingleThreadExecutor() 我预测是500秒,没想到还真是500.306秒
打印日志如下:
当前线程名称pool-3-thread-1
当前线程名称pool-3-thread-1
当前线程名称pool-3-thread-1
... 线程ID 编号取值都是1

三种线程池为什么有这么大的差距?

分别看一下这三种线程池源码的构造方法

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


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

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

构造方法里核心线程数、最大线程数,也就是前2个参数区别(阻塞队列这里我们忽略不考虑)!造成了这个场景的效率差距!

newCachedThreadPool 最大允许开启21亿个线程处理任务


newFixedThreadPool 最大允许开启10个线程处理(我们上方创建对象传入的10)


newSingleThreadExecutor 最大允许开启1个线程处理任务

所以 对应的结果日志时间上的比例也是如此了
第一次 线程直接处理完毕了,而第二次同期时间才处理了10个 剩余990任务都是阻塞的,所以第一次比第二次早了 990 *0.5睡眠的时间 也就是大约48.5秒
第二次 比第三次线程也多了10倍 所以第二次比第三次消耗的时间的10倍。

为什么阿里巴巴开发手册抵触使用Java自带线程池

上面分析了3种Java自带的线程池的效率,发现差距就差距在同时工作的线程数。线程数是CPU的压力。但是如果处理不过来的队列,就会挤压在内存中,就一定有OOM(内存泄漏的问题 Out Of Memory)。

所以就需要使用拒绝策略的线程池。核心就是 BlockingQueue的问题!如果任务来不及处理,且队列满了,此时开始执行拒绝策略,就可以避免OOM的问题!

线程的提交优先级、执行优先级

为什么线程打印的线程ID不按照顺序打印?为什么线程9比线程10执行慢?

当前线程名称pool-2-thread-1
当前线程名称pool-2-thread-2
......
当前线程名称pool-2-thread-7
当前线程名称pool-2-thread-8
当前线程名称pool-2-thread-10
当前线程名称pool-2-thread-9

源码分析

ExecutorService子类AbstractExecutorService 子类实现的 submit()

    public <T> Future<T> submit(Runnable task, T result) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task, result);
        execute(ftask);
        return ftask;
    }

这里可以看到 submit() 调用了execute()

ThreadPoolExecutor IDEA点进去 搜索execute()方法

    /**
     * Checks if a new worker can be added with respect to current
     * pool state and the given bound (either core or maximum). If so,
     * the worker count is adjusted accordingly, and, if possible, a
     * new worker is created and started, running firstTask as its
     * first task. This method returns false if the pool is stopped or
     * eligible to shut down. It also returns false if the thread
     * factory fails to create a thread when asked.  If the thread
     * creation fails, either due to the thread factory returning
     * null, or due to an exception (typically OutOfMemoryError in
     * Thread.start()), we roll back cleanly.
     *
     * @param firstTask the task the new thread should run first (or
     * null if none). Workers are created with an initial first task
     * (in method execute()) to bypass queuing when there are fewer
     * than corePoolSize threads (in which case we always start one),
     * or when the queue is full (in which case we must bypass queue).
     * Initially idle threads are usually created via
     * prestartCoreThread or to replace other dying workers.
     *
     * @param core if true use corePoolSize as bound, else
     * maximumPoolSize. (A boolean indicator is used here rather than a
     * value to ensure reads of fresh values after checking other pool
     * state).
     * @return true if successful
     */
    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;
    }
特别说明 
addwork() 检查是否可以根据当前池状态和给定界限(核心或最大值)添加新的工作线程
workQueue.offer() 如果可以在不违反容量限制的情况下立即将指定元素插入此队列
workerCountOf() 查看当前线程池某状态的线程数
    /**
     * Executes the given task sometime in the future.  The task
     * may execute in a new thread or in an existing pooled thread.
     *
     * If the task cannot be submitted for execution, either because this
     * executor has been shutdown or because its capacity has been reached,
     * the task is handled by the current {@code RejectedExecutionHandler}.
     *
     * @param command the task to execute
     * @throws RejectedExecutionException at discretion of
     *         {@code RejectedExecutionHandler}, if the task
     *         cannot be accepted for execution
     * @throws NullPointerException if {@code command} is null
     */
     public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        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);
    }

上面的代码注释中的3个steps,即是3个if执行的内容,大致含义如下:

由上图得知:线程的执行顺序:核心线程、非核心线程、队列任务