我正在开发一个程序,该程序可以处理大量项目并将每个结果发送到 REST 服务器。该服务器的速率限制为每小时 2000 个请求,因此程序必须在处理两个项目之间暂停一段时间。处理失败的项目应随后呈现给用户。
一切都工作正常,直到我发现程序关闭时 shutdownNow()
不起作用。 UI 关闭,但执行器继续工作。下面是代码的简短摘要。
ExecutorService exec = Executors.newSingleThreadExecutor();
SomeProcessor p = new SomeProcessor();
void process() {
exec.submit(() -> {
Stream<SomeObject> s = ...
List<SomeObject> failed = p.process(s);
// show failed in UI
};
}
void exit() {
exec.shutdownNow();
}
还有 SomeProcessor
类:
List<SomeObject> process(Stream<SomeObject> s) {
List<SomeObject> failed = s
.sequential()
.filter(o -> !ignore(o)) // ignore irrelevant items
.peek(o -> pause()) // make sure not to exceed server's rate limit
.filter(o -> !process(o)) // keep items failed to process
.collect(Collectors.asList());
return failed;
}
void pause() {
try {
TimeUnit.MILLISECONDS.sleep(...);
} catch (final InterruptedException e) {
Thread.interrupted();
}
}
boolean process(SomeObject o) {
if (Thread.interrupted()) // make task interruptible
// *** but then what? ***
try {
// process o and send result to server
return true;
} catch (SomeException e) {
return false;
}
}
我猜 shutdownNow()
不起作用,因为任务不可中断。所以我试图让任务可中断,但我不知道它应该是什么样子。有什么想法吗?
还有一个额外问题。 pause()
方法执行其应该执行的操作。尽管如此,我还是宁愿使用类似 ScheduledThreadPoolExecutor.scheduleAtFixedRate(.)
的东西,但然后处理任务流。有这样的东西存在吗?
感谢您的帮助!
最佳答案
看看你的pause
方法:
void pause() {
try {
TimeUnit.MILLISECONDS.sleep(...);
} catch (final InterruptedException e) {
Thread.interrupted();
}
}
此时您已经检测到中断,但可以通过再次设置线程的中断状态来对其使用react,或者至少尝试这样做,因为 Thread.interrupted()
应该是 Thread.currentThread().interrupt()
来实现这一点,如果你不想支持中断的话,这很好,但在这里它会适得其反。它会导致您的下一个 sleep
调用立即抛出 InterruptedException
,您可以以相同的方式处理该异常,依此类推。因此,您不仅可以继续处理剩余元素,而且无需在元素之间 hibernate 。
当您将方法更改为
void pause() {
try {
TimeUnit.MILLISECONDS.sleep(...);
} catch (final InterruptedException e) {
throw new IllegalStateException("interrupted");
}
}
中断将终止您的流操作并抛出IllegalStateException
。为了清楚起见,您可以为此场景定义自己的异常类型(扩展RuntimeException
),以区别于所有其他异常类型。
关于java - 关闭正在处理 Stream 的 ExecutorService,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/51559478/