高并发接口总被打崩？我用 ArrayBlockingQueue + 底层源码深度剖析搞定流控

一、实现原理⚠️注意✔️有界阻塞队列容量固定必须在初始化时指定长度无自动扩容机制。✔️先进先出FIFO入队元素从队尾添加出队元素从队首取出。✔️存取互斥所有读写操作共享同一把ReentrantLock 锁同一时间只能执行入队或出队操作。二使用场景1、流控例子package com.nl; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; /** * 基于阻塞队列的高并发流量控制 * 作用限制同时进入系统的请求数保护系统不被压垮 */ public class FlowControl { // 队列容量 最大允许并发排队的请求数流控核心 private static final int MAX_QUEUE_SIZE 2; // 阻塞队列线程安全自动实现阻塞/唤醒 private static final BlockingQueueString QUEUE new ArrayBlockingQueue(MAX_QUEUE_SIZE); public static void main(String[] args) { System.out.println(系统启动最大并发排队请求 MAX_QUEUE_SIZE); // 消费者固定线程数处理请求保护系统 // 这里只开 1 个处理线程你可以根据系统能力开 3/5/10 个 new Thread(FlowControl::handleRequest, Consumer-Thread).start(); // 生产者模拟高并发请求10个并发请求 for (int i 1; i 5; i) { final int requestNo i; new Thread(() - { try { // 关键队列满了会自动阻塞等待 → 真正限流 QUEUE.put(Request- requestNo); // 阻塞 1 秒丢弃请求 // boolean success QUEUE.offer(Request- requestNo, 1, TimeUnit.SECONDS); // if (!success) { // System.out.println(请求过多系统繁忙拒绝请求 requestNo); // } System.out.println(【入队】 Thread.currentThread().getName() → Request- requestNo); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }, Producer- i).start(); } } /** * 消费处理请求核心业务 */ private static void handleRequest() { while (true) { try { // 关键队列空会自动阻塞不消耗CPU String request QUEUE.take(); // 模拟业务处理如接口调用、数据库操作 System.out.println(【处理】 Thread.currentThread().getName() 处理 → request); Thread.sleep(500); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } } } }2、输出结果系统启动最大并发排队请求2 【入队】Producer-1 → Request-1 【入队】Producer-4 → Request-4 【处理】Consumer-Thread 处理 → Request-1 【入队】Producer-2 → Request-2 【处理】Consumer-Thread 处理 → Request-4 【入队】Producer-3 → Request-3 【处理】Consumer-Thread 处理 → Request-2 【入队】Producer-5 → Request-5 【处理】Consumer-Thread 处理 → Request-3 【处理】Consumer-Thread 处理 → Request-5⚠️注意✔️阻塞队列是高并发限流、削峰、保护系统的最简单高效方案✔️请求不会冲垮系统CPU / 线程 / 内存都可控三、源码分析1、构造函数public class ArrayBlockingQueueE extends AbstractQueueE implements BlockingQueueE, java.io.Serializable { ...... //数据元素数组 final Object[] items; //下一个待取出元素索引 int takeIndex; //下一个待添加元素索引 int putIndex; //数组元素个数 int count; //ReentrantLock 内部锁 final ReentrantLock lock; //消费者条件 private final Condition notEmpty; //生产者条件 private final Condition notFull; public ArrayBlockingQueue(int capacity) { this(capacity, false); } // public ArrayBlockingQueue(int capacity, boolean fair) { if (capacity 0) throw new IllegalArgumentException(); this.items new Object[capacity]; lock new ReentrantLock(fair); notEmpty lock.newCondition(); notFull lock.newCondition(); } ...... }2、ReentrantLock并发控制// 出队 public E take() throws InterruptedException { // ReentrantLock final ReentrantLock lock this.lock; lock.lockInterruptibly(); try { //count判断 while (count 0) notEmpty.await(); return dequeue(); } finally { lock.unlock(); } } // 入队 public void put(E e) throws InterruptedException { checkNotNull(e); final ReentrantLock lock this.lock; lock.lockInterruptibly(); try { while (count items.length) notFull.await(); enqueue(e); } finally { lock.unlock(); } }锁机制使用ReentrantLock保证线程安全入队和出队操作共用同一把锁实现完全互斥。阻塞等待notEmpty当队列为空count0时出队线程会阻塞在该对象上等待新元素入队。notFull当队列已满countlength时入队线程会阻塞在该对象上等待队列腾出空位。3、数据结构3.1、数组Object[]final Object[] items this.items;3.2、入队和出队// 出队 private E dequeue() { // assert lock.getHoldCount() 1; // assert items[takeIndex] ! null; final Object[] items this.items; SuppressWarnings(unchecked) E x (E) items[takeIndex]; items[takeIndex] null; if (takeIndex items.length) takeIndex 0; count--; if (itrs ! null) itrs.elementDequeued(); notFull.signal(); return x; } // 入队 private void enqueue(E x) { // assert lock.getHoldCount() 1; // assert items[putIndex] null; final Object[] items this.items; items[putIndex] x; if (putIndex items.length) putIndex 0; count; notEmpty.signal(); }⚠️注意✔️底层采用静态数组实现。✔️数组中无元素的位置会被 null 占位因此空间利用率固定。4、双指针设计4.1、入队put/offer从putIndex位置开始添加元素。putIndex自增到达数组末尾时重置为 0循环数组。元素入队成功后唤醒阻塞在notEmpty上的出队线程。4.2、出队take/poll从takeIndex位置开始取出元素。takeIndex自增到达数组末尾时重置为 0循环数组。元素出队成功后唤醒阻塞在notFull上的入队线程。⚠️指针设计✔️putIndex和takeIndex均从队首向队尾循环移动严格保证 FIFO 顺序。✔️ArrayBlockingQueue的双指针putIndex和takeIndex是为了在静态数组上高效实现循环队列环形缓冲区。5、固定长度的静态数组底层是固定长度的静态数组没有扩容机制。用putIndex记录下一个入队元素要存放的位置用takeIndex记录下一个出队元素要取出的位置。两个指针都从 0 开始向后移动到达数组末尾index array.length时重置为 0形成逻辑上的环形复用避免数组空间浪费。三、索引变化逻辑为什么初始都是 0put 和 take 会不会冲突两个索引都从 0 开始put 和 take 不会冲突核心原因是count计数 锁的互斥性保证了安全四、总结优势有界设计避免了内存溢出风险锁和条件变量的组合实现了高效的生产者 - 消费者模式。局限容量固定无法动态扩展独占锁ReentrantLock实现线程安全入队和出队操作使用同一个锁对象在高并发场景下会成为性能瓶颈。