Java 实现阻塞队列

阻塞队列是线程安全的队列核心特性入队阻塞队列满时生产者线程阻塞直到队列有空闲位置出队阻塞队列空时消费者线程阻塞直到队列有元素是生产者-消费者模型的核心组件我会带你实现最经典的有界阻塞队列基于数组锁等待/通知机制这也是 JDKArrayBlockingQueue的核心原理。一、核心实现原理数据结构固定大小的数组存储元素线程安全使用ReentrantLock保证原子操作阻塞唤醒Condition条件队列比wait/notify更精准notEmpty队列非空唤醒消费者notFull队列非满唤醒生产者二、完整代码实现importjava.util.concurrent.locks.Condition;importjava.util.concurrent.locks.ReentrantLock;/** * 基于数组实现的有界阻塞队列 * param E 队列元素类型 */publicclassMyBlockingQueueE{// 存储元素的数组privatefinalObject[]items;// 队头指针取元素用privateinttakeIndex;// 队尾指针放元素用privateintputIndex;// 队列元素数量privateintcount;// 可重入锁保证入队/出队的线程安全privatefinalReentrantLocklock;// 条件队列非空消费者等待/唤醒privatefinalConditionnotEmpty;// 条件队列非满生产者等待/唤醒privatefinalConditionnotFull;/** * 构造方法指定队列容量 * param capacity 队列最大容量 */publicMyBlockingQueue(intcapacity){if(capacity0){thrownewIllegalArgumentException(容量必须大于0);}this.itemsnewObject[capacity];// 默认非公平锁也可传true创建公平锁this.locknewReentrantLock();this.notEmptylock.newCondition();this.notFulllock.newCondition();}/** * 阻塞入队队列满时阻塞线程 * param element 入队元素 * throws InterruptedException 线程中断异常 */publicvoidput(Eelement)throwsInterruptedException{// 加锁保证原子操作lock.lockInterruptibly();try{// 循环判断防止虚假唤醒while(countitems.length){// 队列满生产者阻塞释放锁等待唤醒notFull.await();}// 入队操作enqueue(element);}finally{// 必须释放锁lock.unlock();}}/** * 阻塞出队队列空时阻塞线程 * return 出队元素 * throws InterruptedException 线程中断异常 */publicEtake()throwsInterruptedException{lock.lockInterruptibly();try{// 循环判断防止虚假唤醒while(count0){// 队列空消费者阻塞释放锁等待唤醒notEmpty.await();}// 出队操作returndequeue();}finally{lock.unlock();}}/** * 私有入队方法仅在加锁后调用 */privatevoidenqueue(Eelement){items[putIndex]element;// 指针循环数组环形复用if(putIndexitems.length){putIndex0;}count;// 唤醒一个等待的消费者notEmpty.signal();}/** * 私有出队方法仅在加锁后调用 */privateEdequeue(){Eelement(E)items[takeIndex];items[takeIndex]null;// 帮助GC回收// 指针循环if(takeIndexitems.length){takeIndex0;}count--;// 唤醒一个等待的生产者notFull.signal();returnelement;}// 获取队列元素数量publicintsize(){lock.lock();try{returncount;}finally{lock.unlock();}}}三、关键细节解释1. 为什么用while而不是if必须用循环判断防止虚假唤醒线程未被唤醒却自动唤醒保证队列状态正确。2. 环形数组设计takeIndex和putIndex到达数组末尾后重置为0实现数组空间复用效率更高。3. 锁与条件队列一把ReentrantLock控制所有操作保证线程安全两个Condition精准唤醒入队后唤醒消费者出队后唤醒生产者比Object.wait/notify效率更高不会误唤醒不相关线程。四、测试生产者-消费者模型publicclassBlockingQueueTest{publicstaticvoidmain(String[]args){MyBlockingQueueIntegerqueuenewMyBlockingQueue(5);// 生产者线程持续入队newThread(()-{try{for(inti1;i10;i){queue.put(i);System.out.println(生产者入队i队列大小queue.size());Thread.sleep(500);}}catch(InterruptedExceptione){Thread.currentThread().interrupt();}},生产者).start();// 消费者线程持续出队newThread(()-{try{while(true){Integerelementqueue.take();System.out.println(消费者出队element队列大小queue.size());Thread.sleep(1000);}}catch(InterruptedExceptione){Thread.currentThread().interrupt();}},消费者).start();}}测试效果队列满时生产者自动阻塞队列空时消费者自动阻塞线程安全无数据错乱五、Java 官方阻塞队列生产环境用实际开发直接用 JUC 包下的实现类无需手写ArrayBlockingQueue数组实现有界公平/非公平可选LinkedBlockingQueue链表实现可选有界/无界PriorityBlockingQueue优先级队列无界SynchronousQueue不存储元素直接传递总结阻塞队列核心队列满/空时线程阻塞条件满足时唤醒实现关键ReentrantLock Condition 环形数组手写核心是理解原理生产环境直接用java.util.concurrent官方实现必须用while判断队列状态防止虚假唤醒