java - 关闭正在处理 Stream 的 ExecutorService

Question

我正在开发一个程序，该程序可以处理大量项目并将每个结果发送到 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(.) 的东西，但然后处理任务流。有这样的东西存在吗？

感谢您的帮助!

Answer 1

看看你的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)，以区别于所有其他异常类型。