Java parallelStream 不使用预期的线程数

Question

Java 8 parallelStream 使用的线程似乎比系统属性 java.util.concurrent.ForkJoinPool.common.parallelism 指定的线程多。这些单元测试表明，我使用自己的 ForkJoinPool 使用所需数量的线程处理任务，但使用 parallelStream 时，线程数量高于预期。

import org.junit.Test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import static org.junit.Assert.assertTrue;

public class ParallelStreamTest {

    private static final int TOTAL_TASKS = 1000;

    @Test
    public void testParallelStreamWithParallelism1() throws InterruptedException {
        final Integer maxThreads = 1;
        System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", maxThreads.toString());
        List<Integer> objects = new ArrayList<>();
        for (int i = 0; i < 1000; i++) {
            objects.add(i);
        }

        final AtomicInteger concurrentThreads = new AtomicInteger(0);
        final AtomicInteger taskCount = new AtomicInteger(0);

        objects.parallelStream().forEach(i -> {
            processTask(concurrentThreads, maxThreads); //expected to be called one at the time
            taskCount.addAndGet(1);
        });

        assertTrue(taskCount.get() == TOTAL_TASKS);
    }

    @Test
    public void testMyOwnForkJoinPoolWithParallelism1() throws InterruptedException {
        final Integer threads = 1;
        List<Integer> objects = new ArrayList<>();
        for (int i = 0; i < TOTAL_TASKS; i++) {
            objects.add(i);
        }

        ForkJoinPool forkJoinPool = new ForkJoinPool(1);
        final AtomicInteger concurrentThreads = new AtomicInteger(0);
        final AtomicInteger taskCount = new AtomicInteger(0);

        forkJoinPool.submit(() -> objects.parallelStream().forEach(i -> {
            processTask(concurrentThreads, threads); //expected to be called one at the time
            taskCount.addAndGet(1);
        }));
        forkJoinPool.shutdown();
        forkJoinPool.awaitTermination(1, TimeUnit.MINUTES);

        assertTrue(taskCount.get() == TOTAL_TASKS);
    }

    /**
     * It simply processes a task increasing first the concurrentThreads count
     *
     * @param concurrentThreads Counter for threads processing tasks
     * @param maxThreads Maximum number of threads that are expected to be used for processing tasks
     */
    private void processTask(AtomicInteger concurrentThreads, int maxThreads) {
        int currentConcurrentThreads = concurrentThreads.addAndGet(1);
        if (currentConcurrentThreads > maxThreads) {
            throw new IllegalStateException("There should be no more than " + maxThreads + " concurrent thread(s) but found " + currentConcurrentThreads);
        }

        // actual processing would go here

        concurrentThreads.decrementAndGet();
    }
}

应该只有一个线程用于处理任务，因为 ForkJoinPool 有parallelism=1 和java.util.concurrent.ForkJoinPool.common.parallelism=1。因此，两个测试都应该通过，但 testParallelStreamWithParallelism1 失败：

java.lang.IllegalStateException: 应该有不超过 1 个并发线程，但找到了 2 个

似乎设置 java.util.concurrent.ForkJoinPool.common.parallelism=1 没有按预期工作，同时处理了超过 1 个并发任务。

有什么想法吗？

Answer 1

Fork/Join 池的并行度设置决定了池工作线程的数量，但由于调用者线程，例如主线程也将处理作业，使用公共池时总会多一个线程。这就是为什么default setting of the common pool is “number of cores minus one” 获得的实际工作线程数等于内核数。

使用您的自定义 Fork/Join 池，流操作的调用者线程已经是池的工作线程，因此，将其用于处理作业不会增加实际工作线程数。

必须强调的是，Stream 实现和 Fork/Join 池之间的交互是完全未指定的，因为 Stream 在底层使用 Fork/Join 框架是一个实现细节。不保证更改默认池的属性会对流产生任何影响，也不保证从自定义 Fork/Join 池的任务中调用流操作将使用该自定义池。

Answer 2

也设置这个参数：

    System.setProperty("java.util.concurrent.ForkJoinPool.common.maximumSpares", "0");

这对我有用。显然（虽然没有很好的记录），允许“备用”线程从默认的 ForkJoinPool 中获取工作。

Answer 3

运行这个例子：

  IntStream.rangeClosed(0,9).parallel().forEach((i) -> {
      try {
        System.out.println("id - " + Thread.currentThread().getName());
      } catch (Exception e) {
      }
    });

当您使用参数 java.util.concurrent.ForkJoinPool.common.parallelism=1 时，您会看到类似

id - main
id - main
id - ForkJoinPool.commonPool-worker-1
id - main
id - ForkJoinPool.commonPool-worker-1
id - main
id - ForkJoinPool.commonPool-worker-1
id - main
id - ForkJoinPool.commonPool-worker-1
id - ForkJoinPool.commonPool-worker-1

正如您现在所知，流使用常见的 ForkJoinPool（并行度=1），此外它们还使用当前线程。

Answer 4

您在第一次发布此问题时删除了正确答案，因此我将对其进行阐述和扩展。你的问题在这里： int currentConcurrentThreads = concurrentThreads.addAndGet(1); 在这里：

objects.parallelStream().forEach(i -> {
  processTask(concurrentThreads, maxThreads); //expected to be called one at the time
  taskCount.addAndGet(1);
});

并行流中的每个线程调用processTask。因此，每个都增加concurrentThreads（但由于某种原因，不是 https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/atomic/AtomicInteger.html#incrementAndGet-- ）。由于每个都并行运行，因此它们都在递增 concurrentThreads 之前可以递减它。因此，您当然超出了预期的线程数。