java - 如何实现线程安全的收集器？

Question

我想要类似于 Collectors.maxBy() 的东西，一个获取集合中顶部元素的收集器(maxBy 只获取一个)。

我有一个 Possibility 的流可以用 Integer score(Possibility) 评分的对象方法。

首先我尝试了:

List<Possibity> possibilities = getPossibilityStream()
    .parallel()
    .collect(Collectors.toList());

if(!possibilities.isEmpty()) {
    int bestScore = possibilities.stream()
        .mapToInt(p -> score(p))
        .max()
        .getAsInt();
    possibilities = possibilities.stream()
        .filter(p -> score(p)==bestScore)
        .collect(Collectors.toList());
}

但这样做时，我扫描了该集合 3 次。一次构建它，第二次获得最高分，第三次过滤它，这不是最优的。此外，可能性的数量可能很大 (>10¹²)。

最好的方法应该是在第一次收集时直接获取最高的可能性，但似乎没有内置的收集器来做这样的事情。

所以我实现了自己的 Collector :

public class BestCollector<E> implements Collector<E, List<E>, List<E>> {

    private final Comparator<E> comparator;

    private final Class<? extends List> listImpl ;

    public BestCollector(Comparator<E> comparator, Class<? extends List> listImpl) {
        this.comparator = comparator;
        this.listImpl = listImpl;
    }

    public BestCollector(Comparator<E> comparator) {
        this.comparator= comparator;
        listImpl = ArrayList.class;
    }

    @Override
    public Supplier<List<E>> supplier() {
        return () -> {
            try {
                return listImpl.newInstance();
            } catch (InstantiationException | IllegalAccessException ex) {
                throw new RuntimeException(ex);
            }
        };
    }

    @Override
    public BiConsumer<List<E>, E> accumulator() {
        return (list, e) -> {
            if (list.isEmpty()) {
                list.add(e);
            } else {
                final int comparison = comparator.compare(list.get(0), e);
                if (comparison == 0) {
                    list.add(e);
                } else if (comparison < 0) {
                    list.clear();
                    list.add(e);
                }
            }
        };
    }

    @Override
    public BinaryOperator<List<E>> combiner() {
        return (l1, l2) -> {
            final int comparison = comparator.compare(l1.get(0), l2.get(0));
            if (comparison == 0) {
                l1.addAll(l2);
                return l1;
            } else if (comparison < 0) {
                return l2;
            } else {
                return l1;
            }
        };
    }

    @Override
    public Function<List<E>, List<E>> finisher() {
        return Function.identity();
    }

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

然后:

List<Possibity> possibilities = getPossibilityStream()
    .parallel()
    .collect(new BestCollector<Possibility>((p1, p2) -> score(p1).compareTo(score(p2)));

这在顺序模式下完成工作(没有 .parallel() )但在并行模式下偶尔会在两个地方出现一些异常:

• A java.lang.IndexOutOfBoundsException Index: 0, Size: 0在行中:

  final int comparison = comparator.compare(list.get(0), e);
  

属于 accumulator()方法

我知道它发生在 list.clear() 时在 list.isEmpty() 之间调用和 list.get(0) .

• A java.lang.NullPointerException在 score(Possibility) 方法中，因为可能性是 null .再次涉及同一行:

  final int comparison = comparator.compare(list.get(0), e);
  

我不明白list.get(0)可以返回 null ...

在并行模式下，有时 list.get(0)提出 IndexOutOfBoundsException有时返回 null .

我知道我的代码不是线程安全的，所以我尝试了几种解决方案:

• 添加synchronized在 BestCollector 的所有方法中:public synchronized …
• 使用线程安全的集合而不是 ArrayList : java.util.concurrent.CopyOnWriteArrayList
• 添加synchronized并使用 CopyOnWriteArrayList同时

• 删除 Characteristics.CONCURRENT来自 Set<Characteristics>的 characteristics()方法

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

但我不知道 Characteristics.CONCURRENT在这里表明我的代码是线程安全的或者我的代码将用于并发处理。

但这些解决方案都没有真正解决问题。

---

事实上，当我从特征中删除 CONCURRENT 时，有时会出现 java.lang.IndexOutOfBoundsException: Index: 0, Size: 0但在行中:

final int comparison = comparator.compare(l1.get(0), l2.get(0));

属于 combiner()方法。

但是，accumulator() 引发的异常方法似乎不再出现。

---

@Holger 的回答是正确的。

完整的解决方案是同时更改 combiner()和 characteristics()方法:

@Override
public BinaryOperator<List<E>> combiner() {
    return (l1, l2) -> {
        if (l1.isEmpty()) {
            return l2;
        } else if (l2.isEmpty()) {
            return l1;
        } else {
            final int comparison = comparator.compare(l1.get(0), l2.get(0));
            if (comparison == 0) {
                l1.addAll(l2);
                return l1;
            } else if (comparison < 0) {
                return l2;
            } else {
                return l1;
            }
        }
    };
}

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

Answer 1

您的代码只有一个重大错误:如果您的收集器不是线程安全的，它不应该报告 Characteristics.CONCURRENT，因为这正是声称它是线程安全的。

您必须了解的重要一点是，对于非CONCURRENT 收集器，框架将执行必要的步骤以线程安全但仍有效的方式使用它:

• 对于每个工作线程，将通过 supplier()
获取一个新容器
• 每个 worker 将使用 accumulator() 函数及其自己的本地容器
• combiner() 将在两个工作线程完成工作后使用
• finisher() 将在所有 工作线程完成其工作且所有容器已组合时使用

所以你所要做的就是确保你的供应商在每次调用时真正返回一个新实例，并且所有函数都是无干扰和无副作用的(除了它们作为参数接收的容器之外的任何其他东西)并且，当然，当您的收集器不是并发收集器时，不要报告 Characteristics.CONCURRENT。

这里不需要 synchronized 关键字或并发集合。

---

顺便说一句，(p1, p2) -> score(p1).compareTo(score(p2)) 形式的Comparator 可以使用 Comparator.comparing(p -> score(p)) 或者如果分值是 int:Comparator.comparingInt(p -> score(p)).

---

最后，您的组合器函数不会检查其中一个列表是否为空。这完美地解释了 combiner 中的 IndexOutOfBoundsException 而 accumulator 中的 IndexOutOfBoundsException 是收集器报告的结果 Characteristics.CONCURRENT…

---

了解向 accumulator() 或 combiner() 方法添加 synchronized 关键字并不能保护构造的函数也很重要通过 lambda 表达式。它将保护构造函数实例的方法，而不是函数的代码本身。与内部类相比，无法将 synchronized 关键字添加到实际函数的实现方法中。