寻求可基于FileProcessingRequest总大小阻塞生产者的并发阻塞队列实现

当然有办法实现这个需求，而且完全不需要从零开始写一个并发阻塞队列——我们可以通过**包装现有JDK并发队列+信号量（Semaphore）**的方式，轻松实现基于FileProcessingRequest总size的生产者阻塞逻辑。下面是具体的思路和实现示例：

核心思路：用Semaphore控制总数据容量 #

Semaphore的本质是一个许可计数器，我们可以把"许可数"映射为允许在队列中存储的总字节数（比如2GB就是2L * 1024 * 1024 * 1024个许可）。生产者在放入请求前，必须先获取对应size的许可；消费者处理完请求后，再释放对应size的许可。结合JDK自带的ArrayBlockingQueue（或其他阻塞队列）来存储实际的请求对象，这样既复用了成熟的并发队列实现，又实现了自定义的容量限制逻辑。

具体实现代码 #

我们可以封装一个自定义的阻塞队列类，内部组合ArrayBlockingQueue和Semaphore：

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

public class SizeLimitingBlockingQueue implements BlockingQueue<FileProcessingRequest> {

    private final BlockingQueue<FileProcessingRequest> delegateQueue;
    private final Semaphore sizeSemaphore;

    // 构造方法：传入总容量（字节）和内部队列的初始容量（可选，可根据业务调整）
    public SizeLimitingBlockingQueue(long totalAllowedSizeBytes, int queueCapacity) {
        this.delegateQueue = new ArrayBlockingQueue<>(queueCapacity);
        // 注意：若总容量超过Integer.MAX_VALUE（约2GB），可改用Guava的LongSemaphore，或调整许可单位为KB/MB
        this.sizeSemaphore = new Semaphore((int) totalAllowedSizeBytes);
    }

    @Override
    public void put(FileProcessingRequest request) throws InterruptedException {
        // 先获取对应size的许可，获取不到则阻塞生产者
        sizeSemaphore.acquire(request.getSize());
        try {
            // 放入内部队列存储
            delegateQueue.put(request);
        } catch (InterruptedException e) {
            // 若放入队列时被中断，必须释放已获取的许可，避免许可泄漏
            sizeSemaphore.release(request.getSize());
            throw e;
        }
    }

    @Override
    public FileProcessingRequest take() throws InterruptedException {
        FileProcessingRequest request = delegateQueue.take();
        // 处理完成后释放对应size的许可，让生产者可以继续放入数据
        sizeSemaphore.release(request.getSize());
        return request;
    }

    // 按需实现offer方法（带超时逻辑）
    @Override
    public boolean offer(FileProcessingRequest request, long timeout, TimeUnit unit) throws InterruptedException {
        if (sizeSemaphore.tryAcquire(request.getSize(), timeout, unit)) {
            try {
                if (delegateQueue.offer(request, timeout, unit)) {
                    return true;
                } else {
                    // 队列已满，释放已获取的许可
                    sizeSemaphore.release(request.getSize());
                    return false;
                }
            } catch (InterruptedException e) {
                sizeSemaphore.release(request.getSize());
                throw e;
            }
        }
        return false;
    }

    // 其余BlockingQueue方法可按需实现，建议优先使用put/take/offer等阻塞方法
    @Override
    public boolean add(FileProcessingRequest request) {
        throw new UnsupportedOperationException("建议使用put或offer方法以保证容量控制");
    }

    // 省略其他未实现的方法，可根据业务需求补充
}

关键注意事项 #

• 大容量场景处理：如果总容量超过Integer.MAX_VALUE（约2GB），JDK自带的Semaphore无法直接支持（许可数为int类型），此时可以改用Guava的LongSemaphore，或者将许可单位调整为KB/MB（比如1许可代表1KB，2GB对应2097152个许可）。
• 单个请求超限处理：如果某个FileProcessingRequest的size大于总允许容量，acquire会永久阻塞，建议在put前添加前置检查：

  if (request.getSize() > sizeSemaphore.availablePermits()) {
      throw new IllegalArgumentException("单个请求大小超过队列总容量限制");
  }
  

• 内存溢出风险：即使控制了总size，所有byte[]都存储在堆内存中，需确保JVM堆内存足够（比如设置-Xmx3G），避免OOM。
• 许可泄漏防护：任何可能中断的操作中，都要确保已获取的许可被正确释放，防止生产者永久阻塞。

为什么不用自行编写队列？ #

JDK的ArrayBlockingQueue等并发队列已经经过了严格的并发测试和优化，处理了线程中断、公平性、锁优化等各种边界情况。通过包装的方式复用这些成熟实现，既保证了代码的可靠性，又减少了自行编写并发代码的风险。

内容的提问来源于stack exchange，提问作者VB_