导航：首页 > 网络技术 >

线程池原理分析

发表于：2024-11-26 作者：热门IT资讯网编辑

编辑最后更新 2024年11月26日，CountDownLatch（计数器）CountDownLatch 位于并发包下，利用它可以完成类似于计数器的功能，如果线程 A 需要等待其他 n 个线程执行完毕后才能执行，此时就可以利用 Count

CountDownLatch（计数器）

CountDownLatch 位于并发包下，利用它可以完成类似于计数器的功能，如果线程 A 需要等待其他 n 个线程执行完毕后才能执行，此时就可以利用 CountDownLatch 来实现这个功能，CountDownLatch 是通过一个计数器来实现的，计数器的初始值为线程数量，每当一个线程完成了自己的任务后，计数器的值就会减1，当计数器的值为0时，表示所有线程已经执行完毕，等待线程就可以恢复。

package com.kernel;import java.util.concurrent.CountDownLatch;public class Test001 {    public static void main(String[] args) throws InterruptedException {        CountDownLatch countDownLatch = new CountDownLatch(2);        new Thread(new Runnable() {            @Override            public void run() {                try {                    Thread.sleep(1000);                } catch (InterruptedException e) {                    e.printStackTrace();                }                System.out.println(Thread.currentThread().getName() + ",子线程开始执行...");                countDownLatch.countDown();                System.out.println(Thread.currentThread().getName() + ",子线程结束执行...");            }        }).start();        new Thread(new Runnable() {            @Override            public void run() {                try {                    Thread.sleep(1000);                } catch (InterruptedException e) {                    e.printStackTrace();                }                System.out.println(Thread.currentThread().getName() + ",子线程开始执行...");                countDownLatch.countDown();                System.out.println(Thread.currentThread().getName() + ",子线程结束执行...");            }        }).start();        countDownLatch.await();        System.out.println("其他线程执行完毕");        System.out.println("等待线程执行。。。");    }}

CyclicBarrier（屏障）

CyclicBarrier 在初始化时会传入一个数量，它会记录调用了 await 方法的线程数，只有这个线程数和创建该对象时提供的数量相同时，所有进入线程等待的线程才会被重新唤醒继续执行。

顾名思义，它就像一个屏幕，人来全了才能一块儿过屏障。

CyclicBarrier 还可以提供一个可以传入 Runable 对象的构造，该线程将在一起通过屏障后所有线程唤醒之前被唤醒

package com.kernel;import java.util.concurrent.BrokenBarrierException;import java.util.concurrent.CyclicBarrier;class Write extends Thread {    private CyclicBarrier cyclicBarrier;    public Write(CyclicBarrier cyclicBarrier) {        this.cyclicBarrier = cyclicBarrier;    }    @Override    public void run() {        System.out.println("线程" + Thread.currentThread().getName() + ",正在写入数据");        try {            Thread.sleep(3000);        } catch (InterruptedException e) {            e.printStackTrace();        }        System.out.println("线程" + Thread.currentThread().getName() + ",写入数据成功.....");        try {            cyclicBarrier.await();        } catch (InterruptedException e) {            e.printStackTrace();        } catch (BrokenBarrierException e) {            e.printStackTrace();        }        System.out.println("所有线程执行完毕");    }}public class Test002 {    public static void main(String[] args) {        CyclicBarrier cyclicBarrier = new CyclicBarrier(5);        for (int i = 0; i < 5; i++) {            Write write = new Write(cyclicBarrier);            write.start();        }    }}

Semaphore（信号量）

Semaphore 是一种基于计数的信号量，创建时可以指定一个值，这个值规定了有多少个线程并发执行，执行前申请，执行完毕后归还，超过那个值后，线程申请信号将会被阻塞，知道有其他占有信号的线程执行完成归还信号

Semaphore 可以用来构建一些对象池，资源池之类的，比如数据库连接池

我们也可以创建计数为1的 Semaphore，将其作为一种类似互斥锁的机制，这也叫二元信号量，表示两种互斥状态

package com.kernel;import java.util.Random;import java.util.concurrent.Semaphore;public class Test003 extends Thread {    private String name;    private Semaphore windows;    public Test003(String name, Semaphore windows) {        this.name = name;        this.windows = windows;    }    @Override    public void run() {        int availablePermits = windows.availablePermits();        if (availablePermits > 0) {            System.out.println(name + ":终于轮到我了");        } else {            System.out.println(name + ":**，能不能快点");        }        try {            windows.acquire();        } catch (InterruptedException e) {            e.printStackTrace();        }        System.out.println(name + ":我要XXX，剩下窗口");        try {            Thread.sleep(new Random().nextInt(1000));        } catch (InterruptedException e) {            e.printStackTrace();        }        System.out.println(name + ":我买完了");        windows.release();    }    public static void main(String[] args) {        Semaphore semaphore = new Semaphore(3);        for (int i = 0; i < 5; i++) {            Test003 test003 = new Test003("第" + i + "个人", semaphore);            test003.start();        }    }}

并发队列

ArrayDeque（数组双端队列）
PriorityQueue（优先级队列）
ConcurrentLinkedQueue（基于链表的并发队列）
DelayQueue（延期阻塞队列，阻塞队列实现了BlockingQueue接口）
ArrayBlockingQueue（基于数组的并发阻塞队列）
LinkedBlockingQueue（基于链表的FIFO阻塞队列）
LinkedBlockingDeque（基于链表的FIFO双端阻塞队列）
PriorityBlockingQueue（带优先级的×××阻塞队列）
SynchronousQueue（并发同步阻塞队列）

ConcurrentLinkedDeque（非阻塞式队列）

是一个×××队列，性能高于阻塞式队列，是一个基于链表实现的线程安全队列，不允许 null

add 和 offer 是加入元素的方法，两者之间没有区别

poll 从队列中取出并删除元素

peek 查看队列头元素，不删除

BlockingQueue（阻塞式队列）

是一个有界队列，阻塞队列常用语生产消费者场景

以下情况会产生阻塞：

队列元素满，还往里面存元素

队列元素空，还想从队列中拿元素

基于阻塞队列的生产者消费者模型

package com.kernel;import java.util.concurrent.ArrayBlockingQueue;import java.util.concurrent.BlockingQueue;import java.util.concurrent.TimeUnit;import java.util.concurrent.atomic.AtomicInteger;class ProducerThread implements Runnable {    private BlockingQueue blockingQueue;    private AtomicInteger atomicInteger = new AtomicInteger();    private volatile boolean flag = true;    public ProducerThread(BlockingQueue blockingQueue) {        this.blockingQueue = blockingQueue;    }    @Override    public void run() {        System.out.println("###生产者线程已经启动###");        while (flag) {            String data = atomicInteger.incrementAndGet() + "";            try {                boolean offer = blockingQueue.offer(data, 2, TimeUnit.SECONDS);                if (offer) {                    System.out.println(Thread.currentThread().getName() + ", 生产队列" + data + "成功");                } else {                    System.out.println(Thread.currentThread().getName() + ", 生产队列" + data + "失败");                }                Thread.sleep(1000);            } catch (InterruptedException e) {                e.printStackTrace();            }        }        System.out.println("生产者线程结束");    }    public void stop() {        this.flag = false;    }}class CustomerThread implements Runnable {    private BlockingQueue blockingQueue;    private volatile boolean flag = true;    public CustomerThread(BlockingQueue blockingQueue) {        this.blockingQueue = blockingQueue;    }    @Override    public void run() {        System.out.println("###消费者线程已经启动###");        while (flag) {            String data = null;            try {                data = blockingQueue.poll(2, TimeUnit.SECONDS);                if (data == null) {                    flag = false;                    System.out.println("消费者超过2秒时间未获取到消息.");                    return;                }                System.out.println("消费者获取到队列信息成功,data:" + data);            } catch (InterruptedException e) {                e.printStackTrace();            }        }        System.out.println("消费者进程结束");    }}public class Test006 {    public static void main(String[] args) throws InterruptedException {        BlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);        ProducerThread producerThread = new ProducerThread(blockingQueue);        CustomerThread customerThread = new CustomerThread(blockingQueue);        new Thread(producerThread).start();        new Thread(customerThread).start();        Thread.sleep(1000 * 10);        producerThread.stop();    }}

线程池介绍

线程池其实就是一个可以容纳线程的容器，其中的线程可以反复利用，节省了反复创建、销毁线程的消耗

线程池有什么优点？

降低资源消耗：重复利用已经创建好的线程而节约反复创建、销毁线程的消耗

提高响应速度：众所周期，创建线程不是立马可以使用的，创建好线程之后进入就绪状态，需要经过 CPU 的调度才能进入运行状态，而利用线程池，只要任务来到，线程池有空闲线程，就可以立即作业

提高线程管理性：线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还可以降低系统稳定性，使用线程池可以进行统一分配、调优和监控

线程池的分类：

newCachedThreadPool

创建一个可缓存线程池，可反复回收利用，若任务数大于当然线程数，则继续创建线程

public class Test008 {    public static void main(String[] args) {        // 可缓存线程池（可重复利用）无限大        ExecutorService executorService = Executors.newCachedThreadPool();        for (int i = 0; i < 100; i++) {            executorService.execute(new Runnable() {                @Override                public void run() {                    System.out.println(Thread.currentThread().getName());                }            });        }    }}

newFixedThreadPool

创建一个定长线程，超出线程存放在队列中

public class Test009 {    public static void main(String[] args) {        ExecutorService executorService = Executors.newFixedThreadPool(3);        for (int i = 0; i < 10; i++) {            executorService.execute(new Runnable() {                @Override                public void run() {                    System.out.println(Thread.currentThread().getName());                }            });        }    }}

newScheduledThreadPool

创建一个定长并支持定时及周期性任务执行的线程池

public class Test010 {    public static void main(String[] args) {        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3);        for (int i = 0; i < 10; i++) {            scheduledExecutorService.schedule(new Runnable() {                @Override                public void run() {                    System.out.println(Thread.currentThread().getName());                }                // 表示延迟3秒执行            },3, TimeUnit.SECONDS);        }    }}

newSingleThreadExecutor

创建一个单线程线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序（FIFO、LIFO、优先级）执行

public class Test011 {    public static void main(String[] args) {        ExecutorService executorService = Executors.newSingleThreadExecutor();        for (int i = 0; i < 10; i++) {            executorService.execute(new Runnable() {                @Override                public void run() {                    System.out.println(Thread.currentThread().getName());                }            });        }        executorService.shutdown();    }}

线程池源码分析

corePoolSize 核心线程数（实际运行线程数）
maximumPoolSize 最大线程数（最多可以创建多少个线程）
keepAliveTime 线程空闲超时时间
unit 时间单位
workQueue 缓存队列

提交一个任务到线程池中去，首先判断当前线程数是都小于 corePoolSize，如果小于 corePoolSize，则创建一个新线程来执行任务

如果当前线程数等于corePoolSize，再来任务的话就会将任务添加到缓存队列中

如果缓存队列已满，在判断当前线程是否小于 maximumPoolSize

如果小于 maximumPoolSize，创建线程执行任务，否则，采取任务拒绝策略进行处理

如果当前线程数大于 corePoolSize，并且某线程空闲时间大于 keepAliveTime，线程被终止，直到线程池中的线程数目不大于corePoolSize

如果允许为核心池中的线程设置存活时间，那么核心池中的线程空闲时间超过 keepAliveTime，线程也会被终止

合理配置线程池

IO 密集型：即该任务需要大量的IO，即大量的阻塞，在单线程上运行IO密集型的任务会导致浪费大量的 CPU 运算能力浪费在等待，所以在 IO 密集型任务中使用多线程可以大大的加速程序运行，即时在单核CPU上，这种加速主要就是利用了被浪费掉的阻塞时间，一般 IO 密集型任务线程设置为 2*核心数+1

CPU 密集型：该任务需要进行大量计算，没有 IO 阻塞，CPU 一直在全速运行，CPU 密集任务只有在真正的多核CPU上才可能得到加速（通过多线程），而在单核CPU上，无论你开几个模拟的多线程，该任务都不可能得到加速，因为CPU总的运算能力就那些，一般 CPU 密集型任务线程设置为核心数+1

Callable

Java中，创建线程一般有两种方式，就是继承 Thread 或者实现 Runable 接口，这两种方式的缺点是在线程任务执行完毕后，无法获得执行结果，所以一般使用共享变量或者共享存储区以及线程通信的方式获得结果，Java 中也提供了使用 Callable 和 Future 来实现获取任务结果的操作，Callable 用来执行任务，产生结果，而 Future 用来获得结果

Future 常用方法

V get() 获取异步执行的结果，如果没有结果可用，此方法会阻塞直到异步计算完成

V get(Long timeout, Timeunit unit) 获取异步执行结果，如果没有结果可用，此方法会阻塞，但是会有时间限制，如果阻塞时间超过设定的timeout时间，该方法将抛出异常

boolean isDone() 如果任务执行结束，无论是正常结束或是中途取消还是发生异常，都返回true

boolean isCanceller() 如果任务完成前被取消，则返回true

boolean cancel(boolean mayInterruptIfRunning) 如果任务还没开始就执行该方法将返回 false，如果任务执行过程中调用 cancel(true) 将以中断执行任务的方式试图停止任务，如果停止成功，返回 true，如果执行 cancel(false) 不会对执行的任务产生影响，此时返回 false，当任务完成后调用该方法将返回 false，参数表示是否中断执行进程

Future 模式

去除主函数的等待时间，使得原本需要等待的时间可以处理其他业务，对于多线程，如果线程 A 要等待线程 B 的结果，那么线程 A 没必要等待 B，直到 B 有结果，可以先拿到一个未来的 Future，等 B 有结果是再取真实的结果

模拟 Future

Data

public interface Data {    // 获取子线程执行结果    public String getRequest() throws InterruptedException;}

FutureData

public class FutureData implements Data {    private boolean flag = false;    private RealData realData;    public synchronized void setRealData(RealData realData) {        if (flag)            return;        this.realData = realData;        flag = true;        // 唤醒        notify();    }    @Override    public synchronized String getRequest() throws InterruptedException {        while (!flag) {            // 等待            wait();        }        // 返回结果        return realData.getRequest();    }}

RealData

public class RealData implements Data {    private String result;    public RealData(String data) throws InterruptedException {        System.out.println("正在下载" + data);        Thread.sleep(5000);        System.out.println("下载完毕！");        result = "author:kernel";    }    @Override    public String getRequest() {        return result;    }}

FutureClient

public class FutureClient {    public Data submit(String requestData) {        FutureData futureData = new FutureData();        new Thread(new Runnable() {            @Override            public void run() {                RealData realData = null;                try {                    realData = new RealData(requestData);                    futureData.setRealData(realData);                } catch (InterruptedException e) {                    e.printStackTrace();                }            }        }).start();        return futureData;    }}

测试

public class Test002 {    public static void main(String[] args) throws InterruptedException {        FutureClient futureClient = new FutureClient();        Data request = futureClient.submit("请求参数");        System.out.println("请求发送成功");        // 主线程该干嘛干嘛去        System.out.println("执行其他任务");        // 获取其他线程的结果        String result = request.getRequest();        System.out.println("获取到结果" + result);    }}

执行流程：

首先创建一个向 FutureClient 发送请求，然后 realData 执行下载任务，将结果封装起来，然后获取结果函数时刻监测线程是否拿到结果，如果拿到了，就返回，如果没有拿到，就一直阻塞，设置结果的函数拿到结果会立即唤醒返回结果的函数

重入锁

重入锁，也叫递归锁，指的是同一线程外层函数获得锁之后，内存函数仍有获取该锁的代码，但不受影响

ReentrantLock（显式锁、轻量级锁）和 Synchronized（内置锁、重量级锁）都是可重入锁

读写锁

程序中涉及一些对共享变量的读写操作时，在没有写操作时，多个线程同时读取是没有任何问题的，如果有一个线程正在进行写操作，其他线程就不应该对其进行读或写操作了

读写不同共存，读读可以共存，写写不能共存

public class Test001 {    private volatile Map cache = new HashMap();    private ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock();    private ReadLock readLock = reentrantReadWriteLock.readLock();    private WriteLock writeLock = reentrantReadWriteLock.writeLock();    public void put(String key, String value) {        try {            writeLock.lock();            System.out.println("写入put方法key：" + key + ",value" + value + ",开始");            Thread.sleep(1000);            cache.put(key, value);            System.out.println("写入put方法key：" + key + ",value" + value + ",结束");        } catch (InterruptedException e) {            e.printStackTrace();        } finally {            writeLock.unlock();        }    }    public String get(String key) {        try {            readLock.lock();            System.out.println("读取key："+ key + ",开始");            Thread.sleep(1000);            String value = (String) cache.get(key);            System.out.println("读取key："+ key + ",结束");            return value;        } catch (InterruptedException e) {            e.printStackTrace();            return null;        } finally {            readLock.unlock();        }    }    public static void main(String[] args) {        Test001 test001 = new Test001();        Thread t1 = new Thread(new Runnable() {            @Override            public void run() {                for (int i = 0; i < 10; i++) {                    test001.put("i", i + "");                }            }        });        t1.start();        Thread t2 = new Thread(new Runnable() {            @Override            public void run() {                for (int i = 0; i < 10; i++) {                    test001.get(i + "");                }            }        });        t2.start();    }}