8-ysyThread-阻塞队列容量热更新

阻塞队列容量热更新

业务说明

什么是阻塞队列？

在 Java 中它由接口 BlockingQueue 定义，虽然名称看起来抽象，底层实现却十分灵活，可以基于 数组，也可以使用 单向 或 双向链表 等结构

与普通队列相比，阻塞队列多了两项关键能力：

1. 阻塞插入 ：当队列已满时，执行 put 的线程会被挂起，直到出现空位
2. 阻塞移除 ：当队列为空时，执行 take 的线程同样会被挂起，直到有元素可取

如何实现阻塞队列热更新？

1. 阻塞队列不支持更新容量

在日常线程池调优过程中，我们可能会遇到一个真实的问题：

队列被塞满了，线程池也跑满了，但我又不能轻易重启服务，只是想把队列容量调大点，临时抗一波压力，有没有办法？

• 如果使用的是 LinkedBlockingQueue，可能会发现它的容量是固定的，根本不支持动态调整

/** The capacity bound, or Integer.MAX_VALUE if none */
    private final int capacity;

2. 动态变更阻塞队列场景

阻塞队列动态变更是否有用？

如果阻塞队列容量固定，遇到访问量超过预期，可能很快就会被打满，造成任务拒绝。而如果能在运行时动态调大阻塞队列容量 ，就能临时缓解系统压力，避免雪崩。举例：默认线程池队列容量为 1000。在 访问激增 或 黑产攻击爆发 时，通过 管理平台 将队列扩容至 10000，有效缓冲流量高峰，避免拒绝关键任务。

为什么不直接调大线程池，而是调整阻塞队列容量？这就涉及到两者调整的预期目标：

• 调大线程 ：提高并发执行能力，前提是底层资源能够承受这么大的并发。要不然出问题就是 雪崩 了
• 调大阻塞队列容量 ：增强任务缓冲能力（更多任务排队），可能会慢处理，但是会最终处理。属于是通过空间换高可用一种方案

调参方式	核心作用	风险/代价	适用场景
增加线程池大小	增加并发处理能力	高：CPU竞争、内存压力、线程切换开销	CPU 不敏感、I/O 密集型或延迟敏感场景
增加队列容量	增加任务缓冲能力	低：只是任务排队变多，但延迟可能增加	弹性应对流量突发、避免任务被拒绝或丢失

所以调参的前提一定是基于系统业务模型：

目标	建议调参方式
提高 并发处理速度	增加 线程数（有限度）
增加 任务缓冲能力	增加 队列容量
限制资源使用、防雪崩	缩小线程数和队列容量，合理配置拒绝策略

3. 初步方案：反射修改字段实现扩容

4. 反射方案的风险与改进方向

虽然实现起来不难，但这里有几个重要的点需要说明：

• 高版本开启模块安全检查 后，无法修改 final 字段；使用反射需要添加安全机制
• 修改的是字段，不影响队列中已有元素，也就是队列中已有元素不会丢失。只影响后续 put() 和 offer() 的入队行为限制
• 修改 capacity 只影响后续 put() 和 offer() 的容量检查，不影响已有元素。但若队列已满，阻塞的 put() 线程可能不会自动唤醒，因为 LinkedBlockingQueue 依赖 notFull 条件变量通知
• 依赖 JDK 内部实现：反射方案直接操作 LinkedBlockingQueue 的私有字段 capacity，依赖其内部实现。如果 JDK 版本升级（如从 Java 8 到 17），字段名、锁机制或其他内部实现可能变更，导致代码失效

阻塞队列容量热更新策略下的“坑”

阻塞队列动态更新不止是咱们需要，RabbitMQ 同样需要。在实际运行中，RabbitMQ 的工作线程池会处理来自大量客户端的请求，这些操作的压力往往具有明显的波峰波谷特征 ，例如在早晚高峰、电商秒杀等期间，通过动态扩容任务队列 ，防止因短期流量冲击造成系统雪崩。RabbitMQ 运行过程中，会监控如线程池活跃线程数、队列使用情况、任务阻塞率等指标，达到设置阈值自动进行扩容

RabbitMQ 团队在面临同样需求时，选择了一个更优雅的做法：直接复制并修改LinkedBlockingQueue源码，做成可变容量版本

反射方案的问题

虽然我们可以通过反射手段绕过 final 限制，修改 LinkedBlockingQueue 中的 capacity 字段，但仅仅修改这个字段值并不能真正实现预期行为

1. 队列已满修改容量无效

当队列已满，调用线程正阻塞在 put() 方法上等待空位，此时，我们通过反射动态将 capacity 从 10 修改为 100，期望 unblock 等待线程。但线程仍然阻塞在 notFull.await()，无法及时感知容量变化！

while (count.get() == capacity) {
    notFull.await();  // 阻塞在这里
}

ps：反射只改字段，不会自动触发条件变量通知

2. 容量变小后无法阻塞

假设当前队列已存入 8 个元素，容量为 10。我们通过反射将容量缩小为 5，期望此后入队操作会被阻塞

while (count.get() == capacity) {
    notFull.await();
}

此时 count.get() == 8，capacity == 5，不相等，所以不会阻塞，直接bypass

RabbitMQ 如何解决热更新问题？

VariableLinkedBlockingQueue 是 java.util.concurrent.LinkedBlockingQueue 的一个克隆版本，扩展支持了运行时修改容量的能力 。与 JDK 原生实现不同，它提供了一个公开的 setCapacity(int) 方法，允许我们在队列运行过程中动态调整其容量上限 ，而无需重建队列或重启线程池。

1. setCapacity() 中自动唤醒等待线程

VariableLinkedBlockingQueue 的 setCapacity 方法在修改容量后，主动触发 notFull.signalAll() 来唤醒阻塞线程：

public void setCapacity(int capacity) {
    final int oldCapacity = this.capacity;
    this.capacity = capacity;
    final int size = count.get();
    // 新容量 > 当前队列大小 && 旧容量下已满 or 接近满
    if (capacity > size && size >= oldCapacity) {
        signalNotFull();
    }
}

2. put() 使用 >= 判断

VariableLinkedBlockingQueue 则使用

while (count.get() >= capacity) {
    notFull.await();
}

总结

VariableLinkedBlockingQueue通过非final的capacity字段、改进的put方法（count.get() >= capacity）和 setCapacity 的唤醒逻辑，解决了 LinkedBlockingQueue 无法动态调整容量的问题