使用Parallel Stream阻止所有其他线程

I have a small program that starts 4 threads, each of thread count even number to 20. I am doing a sleep on one of the thread execution only but seems all other threads are blocked waiting the blocked thread to resume.

我有一个小程序,启动4个线程,每个线程计数偶数到20.我只在一个线程执行睡眠,但似乎所有其他线程被阻止等待被阻止的线程恢复。

Why enable the parallelism stream processing is causing such behaviour?

为什么启用并行流处理会导致这种行为?

  public class Program {

        static int first;

        public static void main(String[] a) throws InterruptedException {
            new Program().runTasks();
        }

        private void runTasks() throws InterruptedException {
            int value = 20;
            ExecutorService executorService = Executors.newCachedThreadPool();
            executorService.execute(() -> numberOfEvens(value));
            first++;
            executorService.execute(() -> numberOfEvens(value));
            executorService.execute(() -> numberOfEvens(value));
            executorService.execute(() -> numberOfEvens(value));

            executorService.shutdown();
        }

        private void numberOfEvens(int val) {

            System.out.println(IntStream.range(1, val).parallel().filter(this::isEven).count());
        }

        private void delay(int d) {
            try {
                Thread.sleep(d);
            } catch (InterruptedException ex) {
                // Do Nothing
            }
        }

        private boolean isEven(int n) {
            if (first == 1)
                delay(1000);
            return n % 2 == 0;
        }
    }

2 个解决方案

#1

Your problem is assuming the static remains available to the thread after start.

您的问题是假设启动后线程仍然可用静态。

The first variable may have a value of zero when you create the first lambda but it will have value 1 soon after and almost certainly by the time the first call to delay occurrs.

创建第一个lambda时,第一个变量的值可能为零,但很快就会有值1,几乎可以肯定是第一次调用延迟时间。

This achieves what you want - it prints Delaying 20 times as expected:

这实现了你想要的 - 它按预期打印延迟20次:

public static class Program {

    public static void main(String[] a) throws InterruptedException {
        new Program().runTasks();
    }

    private void runTasks() throws InterruptedException {
        int value = 20;
        ExecutorService executorService = Executors.newCachedThreadPool();
        executorService.execute(() -> numberOfEvens(value, true));
        executorService.execute(() -> numberOfEvens(value, false));
        executorService.execute(() -> numberOfEvens(value, false));
        executorService.execute(() -> numberOfEvens(value, false));
        executorService.execute(() -> numberOfEvens(value, false));
        executorService.execute(() -> numberOfEvens(value, false));
        executorService.execute(() -> numberOfEvens(value, false));
        executorService.execute(() -> numberOfEvens(value, false));
        executorService.execute(() -> numberOfEvens(value, false));
        executorService.execute(() -> numberOfEvens(value, false));
        executorService.execute(() -> numberOfEvens(value, false));
        executorService.execute(() -> numberOfEvens(value, false));
        executorService.execute(() -> numberOfEvens(value, false));

        executorService.shutdown();
    }

    private void numberOfEvens(int val, boolean first) {

        System.out.println(IntStream.range(1, val).parallel().filter(n -> isEven(n, first)).count());
    }

    private void delay(int d) {
        try {
            System.out.println("Delaying");
            Thread.sleep(d);
        } catch (InterruptedException ex) {
            // Do Nothing
        }
    }

    private boolean isEven(int n, boolean first) {
        if (first) {
            delay(1000);
        }
        return n % 2 == 0;
    }
}

#2

You never initialize first so it starts as 0, from sun doc:

你从不初始化,所以从sun doc开始为0:

It's not always necessary to assign a value when a field is declared. Fields that are declared but not initialized will be set to a reasonable default by the compiler. Generally speaking, this default will be zero or null, depending on the data type. Relying on such default values, however, is generally considered bad programming style.

声明字段时并不总是需要分配值。声明但未初始化的字段将由编译器设置为合理的默认值。一般来说,此默认值将为零或null,具体取决于数据类型。然而,依赖于这样的默认值通常被认为是糟糕的编程风格。

So when you are doing first++ it's value is 1 and so all the other threads sleeps.
Initializing first to 1 should solve the problem, like this:

因此,当您第一次执行++时,它的值为1,因此所有其他线程都会休眠。首先初始化为1应解决问题,如下所示:

static int first = 1;

Default Values