java - 当 future 任务中的一个条件满足时，终止正在运行的其他 future 任务

Question

使用java 8

我正在尝试编写一个程序来从差异服务器下载日志文件并在这些日志文件中搜索给定的文本。我现在正在同步进行。我想并行执行，发现可以使用java中的Future来完成。我正在使用 apache.commons.io 从 URL 下载文件。这是代码片段:

ExecutorService executorService = Executors.newCachedThreadPool();
    List<Future<XCluster>> clusterFutures = new ArrayList<>();
    for(XCluster cluster: clusters) {
        clusterFutures.add(executorService.submit(() ->  {
            return downloadAndSearch(textToSearch, cluster);
        }));
    }
    //For now I'm not doing anything with returned value from Future

但现在我想终止在 Future 下启动的其他下载搜索操作，因为给定的搜索预计只能在其中一台服务器中找到。所以没有必要继续我开始的其他 future 任务。任何人都可以建议一种方法来做到这一点吗？我使用的是 java 8，也欢迎其他选项。提前致谢!

Answer 1

ExecutorService 有一个 shutdownNow 方法，它将停止所有线程并关闭服务。

编辑:

我用 shutdownNow 做了一些实验，我发现它不能像我想象的那样停止线程。 AFAIK它使用interrupts()，但并非所有线程都对中断使用react。

所以我能想到的最好的选择是:

首先，创建一个 Indicator 类:

public static class Indicator{

    private boolean isReady = false;

    public void ready(){
        isReady = true;
    }

    public boolean isReady(){
        return isReady;
    }
}

您启动的线程应共享一个 Indicator 实例来进行通信。所以你可以像这样创建一个 Callable:

public static class Processor implements Callable<Integer> {

    private volatile Indicator indicator;

    private Integer number;

    public Processor(Integer integer, Indicator isReady){
        this.number = integer;
        this.indicator = isReady;
    }

    @Override
    public Integer call() throws Exception {
        System.out.println("Thread started:" + Thread.currentThread().getId());
        int counter = 0;
        while (!indicator.isReady &&counter < number) {
           // Make complicated things
            Math.sin(counter);
            counter++;
        }
        if(indicator.isReady){
            //another thread finished
            //delete resources
            System.out.println("Thread interrupted: "  + Thread.currentThread().getId() + " " + counter);
            return -1;
        } else {
            System.out.println("Thread finished: "  + Thread.currentThread().getId() + " " + counter);
            indicator.ready();
            return counter;
        }
    }
}

这样，当第一个线程准备好时，它可以停止其他线程，然后它们自己进行清理。

我尝试如下:

public static void main(String[] args) throws ExecutionException, InterruptedException {
    ExecutorService executorService = Executors.newCachedThreadPool();
    List<Future<Integer>> clusterFutures = new ArrayList<>();
    Indicator indicator = new Indicator();
    clusterFutures.add(executorService.submit(new Processor(100, indicator)));
    clusterFutures.add(executorService.submit(new Processor(10000, indicator)));
    clusterFutures.add(executorService.submit(new Processor(10000000,indicator)));
}

示例输出:

Thread started:11
Thread started:12
Thread finished: 11 100
Thread interrupted: 12 1001
Thread started:13
Thread interrupted: 13 0

旁注:引用的类不必是静态内部类，只是在一个文件中进行实验更容易。