Java並行コード改善のアイデア

Question

私は最近、Java並行処理タスクのコーディング・インタビューを行いましたが、残念ながら仕事を得ていませんでした。最悪の部分は私が最善を尽くしたことですが、今どこが間違っているのか分からなくなっています。誰も私のコードの上で改善できるものについて私にいくつかのアイデアを与えるのを助けることができますか？ありがとう

質問はかなり曖昧です。高レベルでタスクを小さな部分に分割し、各部分を処理し、部分結果を最終結果に結合する4つのジェネリックインターフェースが与えられたら、インターフェースの中央コントローラ部分を実装するように求められます。唯一の要件は、部分的な結果処理で並行処理を使用することです。「コードは生産品質でなければなりません」

私のコードは以下のとおりです（インタフェースが与えられています）。私が正しく理解していれば、私はそうここ

// adding V,W in order to use in private fields types 
public class ControllerImpl<T, U, V, W> implements Controller<T, U> { 

    private static Logger logger = LoggerFactory.getLogger(ControllerImpl.class); 

    private static int BATCH_SIZE = 100; 

    private Preprocessor<T, V> preprocessor; 
    private Processor<V, W> processor; 
    private Postprocessor<U, W> postprocessor; 

    public ControllerImpl() { 
     this.preprocessor = new PreprocessorImpl<>(); 
     this.processor = new ProcessorImpl<>(); 
     this.postprocessor = new PostprocessorImpl<>(); 
    } 

    public ControllerImpl(Preprocessor preprocessor, Processor processor, Postprocessor postprocessor) { 
     this.preprocessor = preprocessor; 
     this.processor = processor; 
     this.postprocessor = postprocessor; 
    } 

    @Override 
    public U process(T arg) { 
     if (arg == null) return null; 

     final V[] parts = preprocessor.split(arg); 
     final W[] partResult = (W[]) new Object[parts.length]; 

     final int poolSize = Runtime.getRuntime().availableProcessors(); 
     final ExecutorService executor = getExecutor(poolSize); 

     int i = 0; 
     while (i < parts.length) { 
      final List<Callable<W>> tasks = IntStream.range(i, i + BATCH_SIZE) 
       .filter(e -> e < parts.length) 
       .mapToObj(e -> (Callable<W>)() -> partResult[e] = processor.processPart(parts[e])) 
       .collect(Collectors.toList()); 
      i += tasks.size(); 

      try { 
       logger.info("invoking batch of {} tasks to workers", tasks.size()); 
       long start = System.currentTimeMillis(); 
       final List<Future<W>> futures = executor.invokeAll(tasks); 
       long end = System.currentTimeMillis(); 
       logger.info("done batch processing took {} ms", end - start); 
       for (Future future : futures) { 
        future.get(); 
       } 
      } catch (InterruptedException e) { 
       logger.error("{}", e);// have comments to explain better handling according to real business requirement 
      } catch (ExecutionException e) { 
       logger.error("error: ", e); 
      } 
     } 

     MoreExecutors.shutdownAndAwaitTermination(executor, 60, TimeUnit.SECONDS); 

     return postprocessor.aggregate(partResult); 
    } 

    private ExecutorService getExecutor(int poolSize) { 
     final ThreadFactory threadFactory = new ThreadFactoryBuilder() 
      .setNameFormat("Processor-%d") 
      .setDaemon(true) 
      .build(); 
     return new ThreadPoolExecutor(poolSize, poolSize, 60, TimeUnit.SECONDS, new LinkedBlockingDeque<>(), threadFactory); 
    } 
} 

Answer 1

削除され、私の仮定を説明するために、コメントの多くに入れて、あなたはTを取り、V []の配列に分割しプリプロセッサを持っていませんでした。次にVをWに変換するプロセッサがあります。そして、W []をUに変換するポストプロセッサー、そうですか？そして、あなたはそれらのものを集める必要があります。

まず第一に、配列とジェネリックは実際には一致しないので、リストではなく配列を返すというのは本当に奇妙です。生産品質のコードでは、汎用配列は使用しないでください。要約するので

、：あなたがリストの代わりに配列を使用する場合は

V[] parts = preprocessor.split(t); 
W[] transformedParts = 
    (W[]) Arrays.stream(parts) // unchecked cast due to the use of generic arrays 
       .parallel() // this is where concurrency happens 
       .map(processor::processPart) 
       .toArray(); 
U result = postProcessor.aggregate(transformedParts); 

、および単一ラインとしてそれを書く：

T --> V1 --> W1 --> U 
     V2 --> W2 
     .  . 
     .  . 
     Vn --> Wn 

だから、あなたがこれを行うことができます

U result = 
    postProcessor.aggregate(
     preprocessor.split(t) 
        .parallelStream() 
        .map(processor::processPart) 
        .collect(Collectors.toList()));