java 8并行流按给定值组织列表

Question

如果元素小于则向左移动，如果更多则向右移动。 java 7风格的简单代码：

private static <T extends Comparable> List<T> doAction(List<T> original, T state) {
    List<T> left = new ArrayList<T>();
    List<T> fight = new ArrayList<T>();
    for (T e : original) {
        if (e.compareTo(state) < 0) {
            left.add(e);
        } else {
            fight.add(e);
        }
    }
    left.addAll(fight);
    return left;
}

如何使用并行流将上面的代码重写为 java 8 流样式？

Answer 1

一种等同于您的原始代码的方法是：

private static <T extends Comparable<T>> List<T> doAction(List<T> original, T state) {
    Map<Boolean, List<T>> collect = original.parallelStream()
        .collect(Collectors.partitioningBy(e -> e.compareTo(state) < 0));
    List<T> left = collect.get(true);
    List<T> right = collect.get(false);
    return Stream.of(left, right).flatMap(List::stream).collect(Collectors.toList());
}

但是，由于您只对数据进行分区，然后再加入它们，一个更简单的解决方案是：

private static <T extends Comparable<T>> List<T> doAction(List<T> original, T state) {
    return original.parallelStream()
        .sorted(Comparator.comparingInt(e -> Integer.signum(e.compareTo(state))|1))
        .collect(Collectors.toList());
}

但请注意，无论哪种情况，您都需要一个相当大的源列表才能从并行执行中受益。

---

请注意，第二种方法甚至可以就地执行，以防原始 List 是可变的（并且您不再需要原始顺序）：

original.sort(Comparator.comparingInt(e -> Integer.signum(e.compareTo(state))|1));

Answer 2

虽然 Holger 的回答很棒，但您（或其他读者）可能有兴趣编写自己的 Collector 或想知道您将如何编写。

首先，您需要一种表示部分结果的方法

static class PartialResult<T extends Comparable<T>>  {
    public List<T> left = new ArrayList<>();
    public List<T> right = new ArrayList<>();
}

现在我们定义自己的Collector，

  static class PartitionCollector<T extends Comparable<T>> implements Collector<T,PartialResult<T>,List<T>>{

    private final T pivot;

    public PartitionCollector(T pivot){
      this.pivot = pivot;
    }

    @Override
    public Supplier<ResultPair<T>> supplier() {
        return ResultPair::new;
    }

    @Override
    public BiConsumer<PartialResult<T>, T> accumulator() {
        return  (result, e) -> {
        if (e.compareTo(pivot) < 0) {
              result.left.add(e);
          }else {
              result.right.add(e);
          }
        };
    }

    @Override
    public BinaryOperator<PartialResult<T>> combiner() {
        BinaryOperator<PartialResult<T>> mergeOp = (r1,r2) ->{
           r1.left.addAll(r2.left);
           r1.right.addAll(r2.right);
           return r1;
        };
        return mergeOp;
    }

    @Override
    public Function<PartialResult<T>, List<T>> finisher() {
        Function<PartialResult<T>,List<T>> finisher = r -> {
            List<T> finalResult = new ArrayList<>(r.left.size() + r.right.size());
            finalResult.addAll(r.left);
            finalResult.addAll(r.right);
            return finalResult;
        };
        return finisher;
    }

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

1. supplier 很简单，我们只需要告诉它创建一个新实例。
2. accumulator，给定一个元素，我们将其放入相应的部分。
3. combiner，我们有两个部分结果（多线程计算了我们解决方案树的两个分支），将它们合并为一个。
4. finsiher，最后，我想要一个完整的列表，而不是部分解决方案。