Java并发代码改进思路

Question

我最近参加了一次关于 Java 并发任务的编码面试，不幸的是没有得到这份工作。最糟糕的是我已经尽力了，但现在我什至不知道哪里出了问题。任何人都可以帮我提供一些关于我可以改进以下代码的想法吗？谢谢

这个问题很模糊。给定 4 个通用接口(interface)，该接口(interface)在较高级别上将任务划分为小块，对每个 block 进行处理并将部分结果组合成最终结果，我被要求实现该接口(interface)的中央 Controller 部分。唯一的要求是在部分结果处理中使用并发，并且“代码必须是生产质量的”

我的代码如下(给出了接口(interface))。我确实发表了很多评论来解释我的假设，但此处已将其删除

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