여러 파일의 병렬 처리 최적화

Question

많은 파일을 처리하는 프로그램이 있습니다. 각 파일에 대해 두 가지 작업을 수행해야합니다. 먼저 파일의 일부를 읽고 처리 한 다음 결과로 MyFileData이됩니다. 저장된. 첫 번째 부분은 병렬화 될 수 있고 두 번째 부분은 병렬화 될 수 없습니다. CPU가 디스크 기다려야합니다 다음 조금 작동하고 또 다른 요청을 발행하고, 다시 대기로 모든 것을 순차적으로 수행

는

은 내가

다음 않았다 ..., 매우 느립니다

class MyCallable implements Callable<MyFileData> { 
    MyCallable(File file) { 
     this.file = file; 
    } 
    public MyFileData call() { 
     return someSlowOperation(file); 
    } 
    private final File file; 
} 

for (File f : files) futures.add(executorService.submit(new MyCallable(f))); 
for (Future<MyFileData> f : futures) sequentialOperation(f.get()); 

그리고 많은 도움이되었습니다. 그러나, 나는 두 가지를 개선하기 위해 싶습니다

• sequentialOperation 먼저 볼 수 있습니다 어떤 결과를 대신 처리의 고정 된 순서대로 실행됩니다. 어떻게 변경할 수 있습니까?

• 처리 할 수천 개의 파일이 있고 수천 개의 디스크 요청을 시작하면 디스크가 손상 될 수 있습니다. Executors.newFixedThreadPool(10)을 사용하여이 숫자를 제한했지만보다 나은 것을 찾고 있습니다. 여러 컴퓨터에서 최적으로 작동하도록 자체 조정해야합니다 (예 : RAID 및/또는 NCQ 등을 사용할 수있는 경우 더 많은 요청을 발행하는 등). HW 구성을 찾는 데 기반 할 수 있다고 생각하지 않지만 처리 속도를 측정하고이를 기반으로 최적화하는 것은 어떻게 든이 가능해야합니다. 어떤 생각?

Answer 1

sequentialOperation 먼저 사용할 어떤 처리 결과 대신 고정 된 순서로 실행 얻는다. 어떻게 변경할 수 있습니까?

정확히 CompletionService의 기능입니다. 즉, 제출 된 명령과 관계없이 작업을 병렬 처리하고 완료된대로 반환합니다.

간체 (테스트하지) 예 :

int NUM_THREADS = Runtime.getRuntime().availableProcessors(); 
ExecutorService executor = Executors.newFixedThreadPool(NUM_THREADS); 
CompletionService<MyFileData> completionService = new ExecutorCompletionService<MyFileData>(executor); 

for (File f : files) futures.add(completionService.submit(new MyCallable(f))); 

for(int i = 0; i < futures.size(); i++) { 
    Future<MyFileData> next = completionService.take(); 
    sequentialOperation(next.get()); 
} 

가 처리 할 파일의 수천은 디스크 요청의 수천을 시작하는 디스크 부수고 발생할 수 있습니다. Executors.newFixedThreadPool (10)을 사용하여이 숫자를 제한했지만보다 나은 것을 찾고 있습니다.

나는 100 % 확실하지 않습니다. 나는 당신이 가지고있는 디스크의 수에 달려 있다고 생각하지만, 디스크 접근 부분이 너무 많은 쓰레드 (디스크 당 하나의 쓰레드가 가능할 것이다)로 분할되어서는 안된다고 생각했을 것이다. 많은 쓰레드가 동시에 하나의 디스크에 접근한다면 , 그것은 독서보다 더 많은 시간을 보냅니다.

Answer 2

sequentialOperation은 먼저 사용 가능한 결과를 처리하는 대신 고정 된 순서로 실행됩니다. 어떻게 변경할 수 있습니까?

가정 :하지만 다른 sequentialOperation과 동시에, 각 someSlowOperation(file); 호출은 시간 변수 시간이 걸릴 것입니다, 따라서, 당신은 빨리 당신이 그것을받을대로 MyFileData을 처리 할.

생성자/소비자 대기열을 설정하여이를 수행 할 수 있습니다.

생산자는 예제에서 실행하는 callables이며, 처리 대기중인 작업 대기열에 결과를 추가하는 추가 비트가 있습니다.

소비자는 sequentialOperation() 호출입니다.이 스레드는 자체 스레드에서 실행되며 하나만 있습니다. 이 모든 쓰레드는 큐의 선두를 차지하고 처리합니다. 프로그램이 끝날 때까지 반복합니다.

이렇게하면 시스템의 모든 리소스 사용을 최대화 할 수 있습니다.

일부 샘플 코드 관련 게시물 : Producer/Consumer threads using a Queue

편집 : 난 당신이 내가 "치이"생각

public class Main { 

    private final ExecutorService producerExecutor = Executors.newFixedThreadPool(10); 
    private final ExecutorService consumerExecutor = Executors.newFixedThreadPool(1); 
    private final LinkedBlockingQueue<MyData> queue = new LinkedBlockingQueue();//or some other impl 

    abstract class Producer implements Runnable{ 
     private final File file; 
     Producer(File file) { 
      this.file = file; 
     } 

     public void run() { 
      MyData result = someLongAssOperation(file); 
      queue.offer(result); 
     } 

     public abstract void someLongAssOperation(File file); 
    } 

    abstract class Consumer implements Runnable { 
     public void run() { 
      while (true) { 
       sequentialOperation(queue.take()); 
      } 
     } 

     public abstract void sequentialOperation(MyData data); 
    } 

    private void start() { 
     consumerExecutor.submit(new Consumer(){ 
      //implement sequentialOperation here 
     }); 

     for (File f : files) { 
      producerExecutor.submit(new Producer(file) { 
       //implement the someLongAssOperation() 
      }); 
     } 

    } 

    public static void main(String[] args) { 
     new Main().start();  
    } 

}