目标是在 Java 8 流的帮助下处理连续的元素流。因此，在处理该流时，将元素添加到并行流的数据源。

Javadoc of Streams在“非干扰”部分描述了以下属性:

For most data sources, preventing interference means ensuring that the data source is not modified at all during the execution of the stream pipeline. The notable exception to this are streams whose sources are concurrent collections, which are specifically designed to handle concurrent modification. Concurrent stream sources are those whose Spliterator reports the CONCURRENT characteristic.

这就是我们尝试使用 ConcurrentLinkedQueue 的原因，它返回 true

new ConcurrentLinkedQueue<Integer>().spliterator().hasCharacteristics(Spliterator.CONCURRENT)

没有明确说明，在并行流中使用时不得修改数据源。

在我们的示例中，对于流中的每个元素，递增的计数器值被添加到队列中，队列是流的数据源，直到计数器大于 N。 通过调用 queue.stream() 在顺序执行时一切正常:

import static org.junit.Assert.assertEquals;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;

public class StreamTest {
    public static void main(String[] args) {
        final int N = 10000;
        assertEquals(N, testSequential(N));
    }

    public static int testSequential(int N) {
        final AtomicInteger counter = new AtomicInteger(0);
        final AtomicInteger check = new AtomicInteger(0);
        final Queue<Integer> queue = new ConcurrentLinkedQueue<Integer>();

        for (int i = 0; i < N / 10; ++i) {
            queue.add(counter.incrementAndGet());
        }

        Stream<Integer> stream = queue.stream();
        stream.forEach(i -> {
            System.out.println(i);

            int j = counter.incrementAndGet();

            check.incrementAndGet();
            if (j <= N) {
                queue.add(j);
            }
        });
        stream.close();
        return check.get();
    }
}

作为第二次尝试，流是并行的并抛出 java.lang.AssertionError，因为检查小于 N 并且不是队列中的每个元素都被处理。流可能提前完成执行，因为队列可能在某个时间点变空了。

import static org.junit.Assert.assertEquals;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;

public class StreamTest {
    public static void main(String[] args) {
        final int N = 10000;
        assertEquals(N, testParallel1(N));
    }

    public static int testParallel1(int N) {
        final AtomicInteger counter = new AtomicInteger(0);
        final AtomicInteger check = new AtomicInteger(0);
        final Queue<Integer> queue = new ConcurrentLinkedQueue<Integer>();

        for (int i = 0; i < N / 10; ++i) {
            queue.add(counter.incrementAndGet());
        }

        Stream<Integer> stream = queue.parallelStream();
        stream.forEach(i -> {
            System.out.println(i);

            int j = counter.incrementAndGet();

            check.incrementAndGet();
            if (j <= N) {
                queue.add(j);
            }
        });
        stream.close();
        return check.get();
    }
}

下一次尝试是在连续流“真正”结束(队列为空)后向主线程发出信号，然后关闭流对象。这里的问题是，流对象似乎只从队列中读取元素一次，或者至少不是连续读取元素，并且永远不会到达流的“真正”末端。

import static org.junit.Assert.assertEquals;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Stream;

public class StreamTest {

    public static void main(String[] args) {
        final int N = 10000;
        try {
            assertEquals(N, testParallel2(N));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static int testParallel2(int N) throws InterruptedException {
        final Lock lock = new ReentrantLock();
        final Condition cond = lock.newCondition();

        final AtomicInteger counter = new AtomicInteger(0);
        final AtomicInteger check = new AtomicInteger(0);
        final Queue<Integer> queue = new ConcurrentLinkedQueue<Integer>();

        for (int i = 0; i < N / 10; ++i) {
            queue.add(counter.incrementAndGet());
        }

        Stream<Integer> stream = queue.parallelStream();
        stream.forEach(i -> {
            System.out.println(i);

            int j = counter.incrementAndGet();

            lock.lock();
            check.incrementAndGet();
            if (j <= N) {
                queue.add(j);
            } else {
                cond.signal();
            }
            lock.unlock();
        });

        lock.lock();
        while (check.get() < N) {
            cond.await();
        }
        lock.unlock();
        stream.close();
        return check.get();
    }
}

由此产生的问题是:

• 我们做错了什么吗？
• 是否是 Stream API 的未指定甚至错误使用？
• 否则我们如何才能实现预期的行为？

最佳答案

“修改 Stream 的源代码不会破坏它”与您的假设“修改将反射(reflect)在正在进行的 Stream 操作中”之间存在显着差异。

CONCURRENT 属性意味着源的修改是允许的，即它永远不会抛出 ConcurrentModificationException，但它不会暗示您可以依赖特定行为来判断是否反射(reflect)了这些更改。

documentation of the CONCURRENT flag本身说:

Most concurrent collections maintain a consistency policy guaranteeing accuracy with respect to elements present at the point of Spliterator construction, but possibly not reflecting subsequent additions or removals.

此 Stream 行为与已知的 ConcurrentLinkedQueue 行为一致:

Iterators are weakly consistent, returning elements reflecting the state of the queue at some point at or since the creation of the iterator. They do not throw ConcurrentModificationException, and may proceed concurrently with other operations. Elements contained in the queue since the creation of the iterator will be returned exactly once.

很难说如何“以其他方式实现所需的行为”，因为您没有以代码以外的任何形式描述“所需的行为”，代码可以简单地替换为

public static int testSequential(int N) {
    return N;
}
public static int testParallel1(int N) {
    return N;
}

因为这是唯一可观察到的效果……考虑 redefining your problem ……

关于java - 即时向 Java 8 并行流添加元素，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/40085379/