使用 ScheduledThreadPoolExecutor 建立有界队列的最佳方法是什么？

Question

Sun Java (1.6) ScheduledThreadPoolExecutor 是 ThreadPoolExecutor 的扩展，内部使用 DelayQueue 的实现，它是一个无界队列。我需要的是一个具有有界队列的ScheduledThreadpoolExecutor，即它对队列中累积的任务有限制，这样当队列中的任务超过限制时，它就会开始拒绝进一步提交任务并防止 JVM 内存不足。

令人惊讶的是，谷歌或 stackoverflow 没有向我指出任何讨论这个问题的结果。有没有我错过的这样的东西？如果没有，我如何实现 ScheduledThreadpoolExecutor 以最佳方式提供我预期的功能？

Answer 1

正如其他人已经指出的那样，没有现成的方法可以做到这一点。只要确保尝试使用“组合”而不是“继承”即可。创建一个新类，该类实现必要的接口，并通过根据需要对必要的方法进行检查来委托给基础 ScheduledThreadPoolExecutor。

您还可以使用此线程中指定的技术 进行简单的修改。您可以使用 Semaphore#tryAcquire 代替使用 Semaphore#acquire，并根据布尔结果决定是否需要调用拒绝处理程序。想想看，我个人认为，库作者直接子类化特定执行器而不是依赖组合来创建普通执行器上的“可调度”包装器是一种疏忽。

Answer 2

如何以不同的方式处理它，即根据队列大小延迟任务提交。执行器服务通过 getQueue() 公开队列。您可以对其调用 size() ，并根据您计划的队列大小限制，您可以开始拒绝任务或开始延迟任务执行（增加计划时间，保持队列大小作为因素之一）。

总而言之，这又不是最好的解决方案；仅供参考，java提供了延迟队列来支持工作窃取。

Answer 3

最简单的解决方法是使用计划执行器仅计划任务，而不实际执行它们。调度程序必须显式检查执行程序队列大小，如果执行程序队列高于阈值，则丢弃任务。

另一种选择是在计划任务中检查 ScheduledThreadPoolExecutor 队列大小。如果队列高于阈值，则立即返回。在这种情况下，任务将立即执行并从队列中删除。所以不会发生溢出。

Answer 4

ScheduledThreadPoolExecutor 不使用队列作为字段，而是调用 getQueue。但它调用 super.getQueue ，它是来自 ThreadPoolExecutor 的队列。您可以使用反射来覆盖它，如下所示：

public class BoundedScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor {
  public BoundedScheduledThreadPoolExecutor(int corePoolSize, RejectedExecutionHandler handler, int queueCapacity) {
    super(corePoolSize, handler);
    setMaximumPoolSize(corePoolSize);
    setKeepAliveTime(0, TimeUnit.MILLISECONDS);
    LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>(queueCapacity) {
            @Override
            public boolean add(Runnable r) {
                boolean added = offer(r);
                if (added) {
                    return added;
                } else {
                    getRejectedExecutionHandler().rejectedExecution(r, CrashingThreadPoolExecutor.this);
                    return false;
                }
            }
        };

    try {
        Field workQueueField = ThreadPoolExecutor.class.getDeclaredField("workQueue");
        workQueueField.setAccessible(true);
        workQueueField.set(this, queue);
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
  }
}

Answer 5

如果您真的，真的不想重新实现ScheduledThreadPoolExecutor，那么您可以扩展它并覆盖所有schedule*方法并实现您的自己的任务界限。但这会相当令人讨厌：

private final Object scheduleMonitor = new Object();

@Override
public ScheduledFuture<?> schedule(Runnable command,
                                   long delay,
                                   TimeUnit unit) 
{
    if (command == null || unit == null)
        throw new NullPointerException();

    synchronized (scheduleMonitor)
    {                
        while (getQueue().size() >= MAX_QUEUE_SIZE)
        {
           scheduleMonitor.wait();
        }
        super.schedule(command, delay, unit);
    }
}

@Override
Runnable getTask() 
{
    final Runnable r = getTask();
    synchronized (scheduleMonitor)
    {
        scheduleMonitor.notify();
    }
    return r;
}

并重复：

• public; ScheduledFutureSchedule(Callable可调用，长延迟，TimeUnit单位)
• public Sc​​heduledFuture ScheduleAtFixedRate（可运行命令，
长初始延迟，
长期,
TimeUnit 单位）
• public Sc​​heduledFuture; ScheduleWithFixedDelay（可运行命令，
长初始延迟，
长时间延迟，
TimeUnit 单位）

注意，这不会阻止重复任务占用队列超过限制，它只会阻止新安排的任务。

另一个警告是，我没有通过在锁定 scheduleMonitor 时调用 super.schedule 来检查任何死锁问题...