Java8 Stream API：如何从非线程安全Consumer强制顺序处理？

首先直接给结论：没办法通过Consumer本身来强制Stream执行顺序处理——Stream的执行模式（顺序/并行）是由Stream自身的配置决定的，和你传入的Consumer是否线程安全完全无关。如果你的Stream是并行模式，哪怕用了非线程安全的Consumer，它依然会在多线程下并发执行，必然导致线程安全问题。

那该怎么解决非线程安全Consumer的问题？有两个明确的方向：

1. 强制Stream以顺序模式运行 #

这是最直接的解决方案，既然Consumer非线程安全，那就干脆不让它并发执行：

• 如果是从集合创建Stream，默认就是顺序流，直接用即可；
• 如果Stream不小心被改成了并行模式（比如调用了parallel()），可以通过sequential()方法强制转回顺序模式。

结合你给出的NonThreadsafeConsumer，示例代码如下：

import java.util.Arrays;
import java.util.List;
import java.util.function.Consumer;

public class StreamExample {
    static class NonThreadsafeConsumer<T> implements Consumer<T> {
        @Override
        public void accept(T arg0) {
            // 这里是你的非线程安全操作，比如修改共享的非线程安全变量
            System.out.println(Thread.currentThread().getName() + " processing: " + arg0);
        }
    }

    public static void main(String[] args) {
        List<String> data = Arrays.asList("apple", "banana", "cherry");
        
        // 正确做法：强制顺序处理
        data.stream()
            .sequential() // 显式调用，即使默认是顺序流，这样写更清晰直观
            .forEach(new NonThreadsafeConsumer<>());
    }
}

运行这段代码你会看到，所有元素都是由同一个线程处理的，完全不会有并发冲突问题。

2. 改造Consumer为线程安全的（如果必须用并行流） #

如果你的数据集极大，并行处理能带来明显的性能提升，那可以把非线程安全的Consumer改成线程安全的。常见的做法是加同步锁，或者使用线程安全的工具类：

static class ThreadsafeConsumer<T> implements Consumer<T> {
    // 用专属锁保证同步操作
    private final Object lock = new Object();

    @Override
    public void accept(T arg0) {
        synchronized (lock) {
            // 原来的非线程安全操作，现在被同步保护
            System.out.println(Thread.currentThread().getName() + " processing: " + arg0);
        }
    }
}

这样即使在并行流中使用，也不会出现线程安全问题，但代价是同步会损失一部分并行带来的性能收益。

注意：别混淆forEachOrdered()的作用 #

很多人会误以为forEachOrdered()能让并行流变成顺序处理，但实际上它只是保证元素的处理结果按Stream的 encounter order输出，元素本身仍然是被并发处理的。如果你的Consumer是非线程安全的，哪怕用了forEachOrdered()，并行流的并发处理依然会引发错误，这点一定要注意！

内容的提问来源于stack exchange，提问作者pvgoddijn