尽早在 ScheduledThreadPoolExecutor 中启动任务/快速启动答案

【问题标题】：Start Tasks in ScheduledThreadPoolExecutor early/ jump start尽早在 ScheduledThreadPoolExecutor 中启动任务/快速启动
【发布时间】：2016-02-01 10:45:09
【问题描述】：

我有一个执行器服务，它定期执行一堆任务。它们在启动时被初始化并经常运行，到目前为止一切顺利。

我现在想添加功能以根据事件快速开始执行这些任务。

我找到了 decorateTask 方法，它允许我存储我安排的任务。但是，我不确定如何让它们运行？

我有覆盖 RunnableScheduledFuture 中的 Delayed 方法以使其在预定义事件上返回 0 的想法，但我也不确定这是否可行，如果我这样做，执行程序会如何表现？

另一个想法是收集所有任务，然后将它们提交以直接在事件上执行。也不确定它会如何表现。

我不能只对它们调用 run，因为它们将在同一个线程中运行。

我希望这一切都有意义。让我知道是否有任何不清楚的地方。

public class EventBasedExecutor extends ScheduledThreadPoolExecutor implements EventBasedExecutorService {
private static final Logger log = Logger.getLogger(EventBasedExecutor.class);

private List<RunnableScheduledFuture<?>> workers = new ArrayList<>();

public EventBasedExecutor(int corePoolSize, ThreadFactory threadFactory) {
    super(corePoolSize, threadFactory);

}

@Override
protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
    workers.add(task);
    return super.decorateTask(runnable, task);
}

@Override
public void executeEarly() {
        // do something here to start the executors work  
}

【问题讨论】：

标签： java multithreading threadpool java.util.concurrent scheduledexecutorservice

【解决方案1】：

最干净且易于理解的方法是收集所有任务并直接在事件上运行它们。我只需要使用 ExecutorService 的 invokeAll 即可。以下代码示例可以提供帮助：

public void handleSomeEvent(Event event) {
    List<Task> tasksToRunOnEvent = getTasksToRunOnEvent(event);
    List<Future<TaskResult> futures = executorService.invokeAll(tasksToRunOnEvent);
    handleTaskResults(futures);
}

【讨论】：

重新提交任务通过执行/调度代码运行，并从旧任务中创建一个新任务（将调用新的装饰等）。我无法复制任务，我需要启动现有实例，以便执行程序无法同时运行它们（因为它们已经运行并且必须重新提交）。
“从旧任务中创建新任务”是什么意思？
我的意思是它接受任务并将其包装在不同的任务中。但是，正确覆盖 decorateTask 方法可以使其正常工作。传递给该方法的 runnable 是上一个任务。只需返回那个，就可以开始正确的任务了。

【解决方案2】：

我听取了 tvelykyy 的建议，玩了一会儿，然后想出了这个 impl：

public class EventBasedExecutor extends ScheduledThreadPoolExecutor implements EventBasedExecutorService {

    private List<RunnableScheduledFuture<?>> workers = new ArrayList<>();

    private int index;

    public EventBasedExecutor(int corePoolSize) {
        super(corePoolSize,  new ThreadFactoryBuilder().setDaemon(true).setNameFormat("message-sender-%d").build());
    }

    @Override
    protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
        if(!workers.contains(runnable)) {
            workers.add(task);
        }
        return super.decorateTask(runnable, task);
    }

    @Override
    public void executeEarly() {
        if(index >= workers.size()) {
            index = 0;
        }

        if(workers.size() == 0) {
            return;
        }

        RunnableScheduledFuture<?> runnableScheduledFuture = workers.get(index);
        index ++;
        execute(runnableScheduledFuture);
    }

}

几句话：

workers.contains(Runnable) 为真，如果您重新提交一个 RunnableScheduledFuture。

你必须通过execute提交它才能让它在线程池中运行（而不是仅仅在runnable上调用run会阻塞当前线程）。

我建议可运行对象实施安全防护，这样它们就不能同时执行两次（对我的用例很重要）。所以基本上：

if( isrunning() ) return;

在 decorateTask 中 - 你必须重新装饰 runnable。这是因为如果必须的话，实现将采用包装好的可运行文件并重新调度它。由于旧的 runnables 已经安排好了，你不想重新安排它们。因此调用 execute 将贯穿修饰并将计划的可运行对象（通常每秒或每隔一段时间运行一次）修饰为不可重复的可运行对象。这意味着，您的计划任务可以运行一次而无需重新计划，同时仍可作为执行程序中的定期运行。

我希望这一切都有意义:)

阿图尔

【讨论】：