작업 훔치기와 병렬 Stream

import java.util.List;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ForkJoinPool;

public final class ParallelSharedCounterBug {
    public static void main(String[] args) throws Exception {
        int[] value = {0};
        CyclicBarrier bothRead = new CyclicBarrier(2);

        try (ForkJoinPool pool = new ForkJoinPool(2)) {
            pool.submit(
                            () ->
                                    List.of(10, 20).parallelStream()
                                            .forEach(
                                                    ignored -> {
                                                        int observed = value[0];
                                                        await(bothRead);
                                                        value[0] = observed + 1;
                                                    }))
                    .get();
        }

        System.out.println("wrong-count=" + value[0]);
    }

    private static void await(CyclicBarrier barrier) {
        try {
            barrier.await();
        } catch (InterruptedException exception) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(exception);
        } catch (BrokenBarrierException exception) {
            throw new IllegalStateException(exception);
        }
    }
}

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.atomic.LongAdder;

public final class WorkStealingDistribution {
    private static final class SumTask extends RecursiveTask<Long> {
        private final int from;
        private final int to;
        private final Map<String, LongAdder> counts;

        private SumTask(int from, int to, Map<String, LongAdder> counts) {
            this.from = from;
            this.to = to;
            this.counts = counts;
        }

        @Override
        protected Long compute() {
            if (to - from <= 4) {
                counts.computeIfAbsent(Thread.currentThread().getName(), key -> new LongAdder())
                        .increment();
                long sum = 0;
                for (int value = from; value < to; value++) {
                    sum += value;
                }
                return sum;
            }
            int middle = from + Math.max(1, (to - from) / 3);
            SumTask left = new SumTask(from, middle, counts);
            SumTask right = new SumTask(middle, to, counts);
            left.fork();
            return right.compute() + left.join();
        }
    }

    public static void main(String[] args) {
        Map<String, LongAdder> counts = new ConcurrentHashMap<>();
        try (ForkJoinPool pool = new ForkJoinPool(4)) {
            long sum = pool.invoke(new SumTask(1, 65, counts));
            System.out.println("sum=" + sum);
        }
        counts.forEach((worker, leaves) -> System.out.println(worker + "=" + leaves.sum()));
    }
}

import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;

public final class ParallelEncounterOrder {
    public static void main(String[] args) {
        List<Integer> source = List.of(5, 1, 4, 2, 3);

        List<Integer> collected = source.parallelStream().map(value -> value * 10).toList();

        ConcurrentLinkedQueue<Integer> completionOrder = new ConcurrentLinkedQueue<>();
        source.parallelStream().forEach(completionOrder::add);

        StringBuilder orderedText = new StringBuilder();
        source.parallelStream().forEachOrdered(value -> orderedText.append(value).append(' '));

        System.out.println("collected=" + collected);
        System.out.println("completion-elements=" + completionOrder.size());
        System.out.println("ordered=" + orderedText.toString().trim());
    }
}

import java.util.EnumSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.BiConsumer;
import java.util.function.BinaryOperator;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collector;

public final class ConcurrentFrequencyCollector {
    private static final class Frequencies
            implements Collector<String, ConcurrentHashMap<String, LongAdder>, Map<String, Long>> {

        @Override
        public Supplier<ConcurrentHashMap<String, LongAdder>> supplier() {
            return ConcurrentHashMap::new;
        }

        @Override
        public BiConsumer<ConcurrentHashMap<String, LongAdder>, String> accumulator() {
            return (map, word) -> map.computeIfAbsent(word, key -> new LongAdder()).increment();
        }

        @Override
        public BinaryOperator<ConcurrentHashMap<String, LongAdder>> combiner() {
            return (left, right) -> {
                right.forEach(
                        (word, count) ->
                                left.computeIfAbsent(word, key -> new LongAdder())
                                        .add(count.sum()));
                return left;
            };
        }

        @Override
        public Function<ConcurrentHashMap<String, LongAdder>, Map<String, Long>> finisher() {
            return mutable ->
                    mutable.entrySet().stream()
                            .collect(
                                    java.util.stream.Collectors.toUnmodifiableMap(
                                            Map.Entry::getKey, entry -> entry.getValue().sum()));
        }

        @Override
        public Set<Characteristics> characteristics() {
            return EnumSet.of(Characteristics.CONCURRENT, Characteristics.UNORDERED);
        }
    }

    public static void main(String[] args) {
        Map<String, Long> counts =
                java.util.stream.Stream.of("java", "stream", "java", "fork", "stream", "java")
                        .parallel()
                        .collect(new Frequencies());
        System.out.println(counts);
    }
}