프로세스 M Java에서 N 개의 스레드에 대한 계산 속도가 느림

Question

느린 계산 (N은 상당히 큰 숫자)을 실행해야하며 속도가 느린 계산에는 많은 IO 대기 시간이 걸리기 때문에 M 스레드에서 수행해야합니다. 나는 모든 계산이 성공한 경우에 잘 작동하는 작은 예제를 작성했습니다. 그러나 계산에 실패하면 원하는 계산을 더 이상 처리하지 않는 것이 좋습니다. 각각의 성공적인 계산은 이미 결과를 데이터베이스에 작성 했으므로 어떤 계산이 실패했는지 결정하고 아직 시작되지 않은 계산을 중지해야합니다.

내 접근 방식은 Executors.newFixedThreadPool에 대한 ExecutorService 인터페이스를 사용하는 것입니다. 그러나 계산 결과 중 하나가 실패 (내 예에서는 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의 작업을 구현합니다. 여전히 콜백 솔루션을 살펴보고 다른 제안을 기대하십시오. I outline in another answer, 당신은 실패를 통지하고, cancel 제출 된 모든 작업을 할 수 콜백을 사용

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 1

:

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 2

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; 
      } 
    }  
} 

Answer 3

당신은 Callable 프레임 워크의 Future면을 통합 할 수 있습니다. (예를 들어, Callback 구현 클래스는 각 Future이 추가 된 Collection에 대한 참조를 가질 수 있습니다.) 완료된 (또는 인수의 값에 따라 시작된) 태스크의 경우 아무 것도 수행하지 않습니다. 나머지는 절대로 시작되지 않습니다.