在Java中处理N个线程上的M个慢速计算

Question

我需要运行 N 个慢速计算（其中 N 是一个相当大的数字）并且希望在 M 个线程上执行此操作，因为慢速计算有大量的 IO 等待时间。我整理了一个小例子，它适用于所有计算都成功的情况。但是，如果计算失败，则期望的行为是停止处理进一步的计算。每个成功的计算都已经将其结果写入数据库，所以我只需要确定哪个计算失败并停止尚未开始的计算。

我的方法是使用 ExecutorService 接口到 Executors.newFixedThreadPool。但是，我没有看到一种明确的方法来识别其中一个计算失败（在我的示例中返回 false）并停止已提交给 ExecutorService 但尚未从池中分配线程的计算。

有没有一种干净的方法可以做到这一点？有更好的方法供我考虑吗？

import java.util.*;
import java.util.concurrent.*;

class Future
{
    static private class MyWorker implements Callable
    {   
        private Integer item;
        public MyWorker(Integer item)
        {
            this.item = item;
        }

        public Boolean call() throws InterruptedException
        {
            if (item == 42) 
            {
                return false;
            }
            else
            {
                System.out.println("Processing: " + item.toString() + " on " + Thread.currentThread().getName());
                Thread.sleep(1000);
                return true;
            }
        }   
    }

    static int NTHREADS = 2;

    public static void main(String args[]) 
    {
        Queue<Integer> numbers = new LinkedList<Integer>();     
        for (int i=1; i<10000; i++)
        {
            numbers.add(i);
        }

        System.out.println("Starting thread test.");

        ExecutorService exec = Executors.newFixedThreadPool(NTHREADS);

        for (Integer i : numbers)
        {
            MyWorker my = new MyWorker(i);
            System.out.println("Submit..." + i.toString());
            exec.submit(my);
            System.out.println("... Done Submit");
        }

        exec.shutdown();

        System.out.println("Exiting thread test.");

    }
}

编辑：这是 afk 建议的有效实现。还是打算看看回调解决方案，希望有其他建议。

import java.util.*;
import java.util.concurrent.*;

class MyFuture
{
    static private class MyWorker implements Callable
    {   
        private Integer item;
        public MyWorker(Integer item)
        {
            this.item = item;
        }

        public Boolean call() 
        {
            if (item == 42) 
            {
                return false;
            }
            else
            {
                System.out.println("Processing: " + item.toString() + " on " + Thread.currentThread().getName());
                try
                {
                    Thread.sleep(1000);
                }
                catch (InterruptedException ie) 
                { 
                // Not much to do here except be grumpy they woke us up...
                } 
                return true;
            }
        }   
    }

    static int NTHREADS = 4;

    public static void main(String args[]) throws InterruptedException
    {
        Queue<Integer> numbers = new LinkedList<Integer>();     
        for (int i=1; i<100; i++)
        {
            numbers.add(i);
        }

        System.out.println("Starting thread test.");

        ExecutorService exec = Executors.newFixedThreadPool(NTHREADS);

        List<Future<Boolean>> futures = new LinkedList<Future<Boolean>>();

        for (Integer i : numbers)
        {
            MyWorker my = new MyWorker(i);
            System.out.println("Submit..." + i.toString());
            Future<Boolean> f = exec.submit(my);
            futures.add(f);
            System.out.println("... Done Submit");
        }

        boolean done = false;

        while (!done)
        {
            Iterator<Future<Boolean>> it = futures.iterator();

            while (it.hasNext()) 
            {
                Future<Boolean> f = it.next();
                if (f.isDone())
                {
                    try
                    {
                        System.out.println("CHECK RETURN VALUE");
                        if (f.get()) 
                        {
                            it.remove();
                        }
                        else
                        {                   
                            System.out.println("IMMEDIATE SHUTDOWN");
                            exec.shutdownNow();
                            done = true;
                            break;
                        }
                    }
                    catch (InterruptedException ie)
                    {
                    }
                    catch (ExecutionException ee)
                    {
                    }
                }
            }
            Thread.sleep(1000);
            if (futures.size() == 0)
            {
                done = true;
            }
        }

        exec.shutdown();

        System.out.println("Exiting thread test.");

    }
}

Answer 1

使用回调，I outline in another answer, 可以通知您失败，cancel 所有提交的作业。 （在我的示例中，Callback 实现类可以引用一个Collection，每个Future 都添加到该Future。）对于那些已经完成（或开始，取决于参数的值）的任务@987654327 @ 什么也没做。其余的永远不会开始。

Answer 2

您可以合并Callable 框架的Future 方面：

 Set futures = new HashSet<Future<Boolean>>
 for (Integer i : numbers)
 {
    MyWorker my = new MyWorker(i);
    System.out.println("Submit..." + i.toString());
    Future<Boolean> f = exec.submit(my);
    futures.add(f);
    System.out.println("... Done Submit");
  }

  for (Future f : futures) {
    if (!f.get().booleanValue()) {
      exec.shutdown();
  }

Answer 3

class Stopper
{
    boolean stopped = false;
    ExecutorService exec;

    public void stop() { if (!stopped) { stopped = true; exec.shutdown(); } }
}

static private class MyWorker implements Callable
{   
    private Integer item;
    private Stopper stopper;
    public MyWorker(Integer item, Stopper stopper)
    {
            this.stopper = stopper;
            this.item = item;
    }

    public Boolean call() throws InterruptedException
    {
            if (item == 42) 
            {
                    stopper.stop();
                    return false;
            }
            else
            {
                    System.out.println("Processing: " + item.toString() + " on " + Thread.currentThread().getName());
                    Thread.sleep(1000);
                    return true;
            }
    }       
}