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，對吧？你必須組裝這些東西。

首先，數組和仿製藥確實不匹配在一起，所以它是真正奇怪的那些方法返回數組，而不是名單。對於生產質量的代碼，不應使用通用數組。

因此，回顧一下：

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

所以，你可以這樣做：如果您使用列表，而不是陣列

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

，並把它寫成一行：

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