多线程(五) ~ 阻塞队列与线程池

景画

发布于 2025-12-19 12:37:05

600

一. 阻塞队列

(一) 定义

什么是阻塞队列? 答: 阻塞队列是⼀种特殊的优先级队列. 也遵守 “先进先出” 的原则, 阻塞队列也有两个特性, 当队列满的时候, 继续⼊队列就会阻塞, 直到有其他线程从队列中取走元素, 当队列空的时候, 继续出队列也会阻塞, 直到有其他线程往队列中插入元素
为什么要用阻塞队列? 答: 当客户端与服务器直接交互时, 如果遇到用户请求高峰期, 大量的请求会直接涌入服务器, 直接处理大量请求数据可能会把服务器搞挂掉, 那么这时就需要一个缓冲器来缓解这种特殊情况

(二) 生产者消费者模型

答: 生产者消费者模型是一种经典的多线程同步协作模式, 通过一个缓冲器来实现生产者与消费者的解耦合, 生产者产出结果后不用再等待消费者出现, 而是直接放到缓冲器里面, 消费者也不从生产者那边拿结果, 而是直接从缓冲器里面取, 这里面的缓冲器就是阻塞队列

(三) 自定义实现阻塞队列

阻塞队列是可以循环使用的且不能扩容的, 我们把它设计成循环队列最核心的就是任务入队列 put 与 任务出队列 take

void put(T task): 当队列满的时候不能继续添加任务, 需要阻塞, 调用wait等待, 同时要搭配锁进行使用, 可以把整体套在synchronized里面, 判断等待的语句应该是while循环, 因为万一存在错误唤醒, 那么循环则有时一层判断保护
T take( ): 队列空时不能继续取出任务, 方法阻塞, 同样需要在锁synchronized中用while语句判断等待, 防止错误唤醒

package test;

import java.util.concurrent.BlockingQueue;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: ran
 * Date: 2025-08-03
 * Time: 21:38
 */
public class MyBlockingQueue<T> {
    private Object[] array ;
    private static final int CAPACITY = 1000;
    private static Object locker = new Object();
    private int head;
    private int tail;
    private int size;

    public MyBlockingQueue(int init) {
        if (init <= 0) {
            array = new Object[CAPACITY];
        }else {
            array = new Object[init];
        }
    }
    // 循环队列
    public void put(T task) throws InterruptedException {
        synchronized (locker) {
            while (size >= array.length) {
                locker.wait();
            }
            array[tail++] = task;
            tail = tail % array.length;
            size++;
            locker.notify();
        }
    }
    public T take() throws InterruptedException {
        synchronized (locker) {
            while (size == 0) {
                locker.wait();
            }
            T task = (T)array[head];
            array[head++] = null;
            head = head % array.length;
            size--;
            locker.notify();
            return task;
        }
    }

    public static void main(String[] args) {
        MyBlockingQueue<Integer> blockingQueue = new MyBlockingQueue<Integer>(2000);
        Thread thread1 = new Thread(() -> {
            int count = 0;
            while (true) {
                try {
                    blockingQueue.put(count++);
                    System.out.println("生产元素: " + count);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        Thread thread2 = new Thread(() -> {
            while (true) {
                try {
                    int count = blockingQueue.take();
                    System.out.println("消费元素: " + count);
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        thread1.start();
        thread2.start();
    }
}

二. 线程池

(一) 定义

什么是线程池? 答: 线程池是一种线程管理的机制, 它会事先创建一批线程放到池子中, 当有任务加入进来时那么就会取出池子中空闲的线程来去执行, 执行完毕后线程不会销毁, 会又回到池子中等待新任务的加入
为什么要用线程池/线程池的优点? 答: 首先我们要知道, 最开始引入线程的概念就是因为进程太重, 频繁创建与销毁进程会消耗大量CPU资源, 所以诞生了线程(轻量级进程), 随着技术发展与互联网普及我们对性能要求越来越高, 渐渐地线程的多次创建与销毁导致的资源消耗我们也不再能接受, 这时候大佬就想了, 既然频繁的创建线程消耗这么大, 那我提前创建好一批线程, 用的时候再取出来, 不用的时候闲置, 反正早晚用到, 只是时间早与时间晚的问题而已, 程序结束也就自动销毁, 这不就节省了大量的开销嘛? 于是线程池应运而生, 而线程池最大的优点就是节省了频繁创建销毁线程的资源消耗…

(二) 线程池参数参数介绍

corePoolSize: 核心线程数量, 也可以理解为最少线程数量, 伴随着线程池的整个生命周期, 不会被销毁, 一个线程池中的线程个数最少不能少过这个数字, 可以为0, 但是不能为负数, 相当于CPU的默频
maximumPoolSize: **最大线程数量, 也可以理解为最多能调用的线程数量, 当任务较多时核心线程不足以应对情况, 那么会创建一些临时线程来辅助执行任务, 但是长时间不干活会被销毁**, 相当于CPU的睿频
keepAliveTime: 临时线程允许的最大空闲时间, 当一波任务高峰需求过去后, 临时线程一直闲着没事干那么继续存在也就没有了意义, 这时当临时线程达到最大空闲时间就会被自动销毁, 节省线程池的维持开销 ↓下方图片解释jdk17线程池的前三个参数源码↓

unit: 它是一个枚举类型, 用来设置keepAliveTime的时间单位, 可以是纳秒 NANOSECONDS| 微妙 MICROSECONDS| 毫秒 MILLISECONDS| 秒 SECONDS| 分 MINUTES| 时 HOURS| 天 DAYS
**workQueue:**传递任务的阻塞队列, 上面已经为大家介绍, 这里不过多赘述
**threadFactory:**线程创建工厂, 不同的线程工厂创建出来的线程的对一些属性进行了不同的初始化设置
**RejectedExecutionHandler:**拒绝策略, 当任务量超出符合, 已有的线程甚至是最大线程数都忙不过来时, 这时候就要学会拒绝了~ ↓接下来下面图片来介绍jdk17源码的四种拒绝策略↓

①:AbortPolicy(): 超过负荷, 直接抛出异常.

②:CallerRunsPolicy(): 调⽤者负责处理多出来的任务, 即哪个线程调用的线程池那么再把任务返还给它, 让它自己执行该任务

③:DiscardOldestPolicy(): 丢弃队列中最⽼的任务

④:DiscardPolicy(): 丢弃新来的任务.

(三) 自我实现线程池

线程池核心方法是 submit, 提交任务, 首先要先定义一个阻塞队列BlockingQueue, 定义submit方法向队列添加任务, 线程池的构造方法创建多个线程, 并且每个线程循环从队列中取出任务, 并执行任务

package test;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: ran
 * Date: 2025-08-03
 * Time: 20:26
 */
public class MyThreadPool {
    BlockingQueue<Runnable> blockingQueue ;
    public MyThreadPool(int n) {
        blockingQueue = new ArrayBlockingQueue<>(100);
        for (int i = 0; i < n; i++) {
            Thread thread = new Thread(() -> {
                try {
                    while (true) {
                        Runnable task = blockingQueue.take();
                        task.run();
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            });
            thread.start();
        }
    }
    public void submit(Runnable task) throws InterruptedException {
        blockingQueue.put(task);
    }

    public static void main(String[] args) throws InterruptedException {
        MyThreadPool threadPool = new MyThreadPool(10);
        for (int i = 0; i < 100; i++) {
            int id = i;
            threadPool.submit(() -> {
                System.out.println("线程名: " + Thread.currentThread().getName() + " 任务ID: " + id);
            });
        }
    }
}

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2025-08-04，如有侵权请联系 cloudcommunity@tencent.com 删除

服务器