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

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()），但在并行模式下，偶尔会在两个地方出现一些异常：

• 一行中的java.lang.IndexOutOfBoundsException Index: 0, Size: 0：

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

accumulator() 方法

我知道在list.isEmpty() 和list.get(0) 之间调用list.clear() 时会发生这种情况。

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

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

我不明白list.get(0) 如何返回null...

在并行模式下，有时list.get(0) 会引发IndexOutOfBoundsException，有时会返回null。

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

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

• 从characteristics()方法的Set<Characteristics>中删除Characteristics.CONCURRENT

  @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() 获取一个新容器
• 每个工作人员都将使用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 的结果...

---

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