Hazelcast ScheduledExecutor 任务队列正在填满答案

【问题标题】：Hazelcast ScheduledExecutor Task Queue is filling upHazelcast ScheduledExecutor 任务队列正在填满
【发布时间】：2020-01-02 03:50:14
【问题描述】：

我正在向 Hazelcast IScheduledExecutorService (Hazelcast 3.12.4) 提交任务

hazelcast.getScheduledExecutorService(DELAYED_EVENT_EXECUTOR_SERVICE_NAME).schedule(task, delay, units);

一直运行良好，现在突然出现错误

java.util.concurrent.RejectedExecutionException: Maximum capacity (200) of tasks reached, for scheduled executor (delayedEventExecutor). Reminder that tasks must be disposed if not needed.
        at com.hazelcast.scheduledexecutor.impl.ScheduledExecutorContainer.checkNotAtCapacity(ScheduledExecutorContainer.java:337)
        at com.hazelcast.scheduledexecutor.impl.ScheduledExecutorContainer.schedule(ScheduledExecutorContainer.java:92)
        at 

com.hazelcast.scheduledexecutor.impl.operations.ScheduleTaskOperation.run(ScheduleTaskOperation.java:45)

像this 这样的其他问题似乎表明我们需要在返回的未来上调用future.dispose()。但是任务在未来执行（这就是重点） - 我们是否需要在某种列表中保留所有期货并在它们执行后dispose()它们？这似乎很疯狂。我想我们可以给任务一个Future的引用，然后在run()方法的最后调用dispose()，但这看起来很傻（更不用说执行器服务是分布式的，任务可能需要被序列化并移动到另一个节点）。

我们如何在未来几个小时内执行的任务的Futures 上调用dispose()？还是有其他方法可以阻止此错误消息？

【问题讨论】：

标签： hazelcast

【解决方案1】：

IScheduledExecutorService 创建并执行在给定延迟后启用的一次性操作。它有两个与您的问题相关的配置属性：

pool-size：执行器每个成员的执行器线程数。
capacity：调度程序每个分区可以拥有的最大任务数。用户应通过调用future.dispose() 来释放任务以释放容量。

你可以在这里找到文档：https://docs.hazelcast.org/docs/latest/manual/html-single/#scheduled-executor-service

将IScheduledFuture 对象保存在列表中并在以后处理它们只会占用容量。相反，您可以调用future.get() 来获取结果并在执行后立即处理它，并在需要时将结果保存在列表中。

为了说明：

ArrayList resultList = new ArrayList<Integer>();
IScheduledExecutorService executorService = member.getScheduledExecutorService("myScheduler");
IScheduledFuture<Integer> future = executorService.schedule(
                new SampleTask(), 10, TimeUnit.SECONDS);

int result= future.get(); // Block until we get the result
resultList.add(result);
future.dispose(); // Always dispose futures that are not in use any more, to release resources

与计划任务关联的期货总是被强引用，以便了解丢失的分区和/或成员，充当本地成员/客户端的侦听器。根据分区特定的调度，未来的任务既存储在主分区中，也存储在它的 N 个备份中。如果 Hazelcast 成员丢失，任务将在备份上重新安排。在成员特定的调度下，未来的任务只存储在成员本身，这意味着如果失去成员，任务也会丢失。

如果您不想在获得结果之前使用future.get() 阻止，请考虑查看ICompletableFuture

【讨论】：

在得到结果之前我不想阻塞 - 如果我想要，我不需要 ScheduledExecutor，我只需同步运行任务。另外，如果我将任务安排在 3 小时内执行，我会阻塞线程 3 小时吗？我的服务器会崩溃。此外，这些任务是可序列化的，并且实际上可能在不同的节点上执行。

【解决方案2】：

这种代码对我不起作用：

IScheduledExecutorService executorService = member.getScheduledExecutorService("myScheduler");
IScheduledFuture<Integer> future = executorService.schedule(
                new SampleTask(), 3, TimeUnit.HOURS);

int result = future.get(); // Block the calling thread for 3 hours and crash the server
future.dispose();

所以我的解决方案是构建一个基于数据库的延迟执行器。我将 Spring Bean 名称、方法名称和序列化参数以及交付时间戳存储在数据库中。丑陋的部分是我必须定期轮询数据库以查找需要传递的事件。如果他们的时间到了，我使用applicationContext.getBean() 调用来查找实例，反序列化参数并使用反射来调用Method。成功后，我删除数据库数据。

【讨论】：