java - 如何使用多个线程并行运行来执行非常繁重的任务

Question

我需要执行一个包含四个步骤的任务。每一步都依赖于前一步的结果。在一个线程中执行所有步骤需要很长时间。我想使用四个线程，每个线程执行一个步骤，并在相邻两个步骤之间使用缓冲区来存储上一步的结果。我正在使用Java在Android平台上进行开发。谁能给我举个例子吗？

非常感谢。

YL

Answer 1

我很好奇，(非神奇并行)代码在 java9 中使用响应式(Reactive)流会是什么样子。事实证明，Java9 基础设施是支离 splinter 的，并且是 used with caution 。所以我选择将示例基于 JavaRx ，提供“流量”。有一个android extension可用于此。

我将其与 Streams、parallelStreams 和顺序流结合起来。

public class FlowStream {

@Test
public void flowStream() {
    int items = 10;

    List<Integer> source = IntStream.range(0, items - 1).boxed().collect(Collectors.toList());

    print("\nstream");
    source.stream().map(this::exp).map(this::exp).forEach(i -> print("streamed %d", i));

    print("\nparallelStream");
    source.parallelStream().map(this::exp).map(this::exp).forEach(i -> print("streamed %d parallel", i));

    print("\nflow");
    Flowable.range(0, items)
            .map(this::exp)
            .map(this::exp)
            .forEach(i -> print("flowed %d", i));

    print("\nparallel flow");
    Flowable.range(0, items)
            .flatMap(v ->
                    Flowable.just(v)
                            .subscribeOn(Schedulers.computation())
                            .map(this::exp)
            )
            .flatMap(v ->
                    Flowable.just(v)
                            .subscribeOn(Schedulers.computation())
                            .map(this::exp)
            ).forEach(i -> print("flowed parallel %d", i));

    await(5000);

}

private Integer exp(Integer i) {
    print("making %d more expensive", i);
    await(Math.round(10f / (Math.abs(i) + 1)) * 50);
    return i;
}

private void await(int i) {
    try {
        Thread.sleep(i);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
}

private void print(String pattern, Object... values) {
    System.out.println(String.format(pattern, values));
}

}

    <!-- https://mvnrepository.com/artifact/io.reactivex.rxjava2/rxjava -->
    <dependency>
        <groupId>io.reactivex.rxjava2</groupId>
        <artifactId>rxjava</artifactId>
        <version>2.2.13</version>
    </dependency>