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

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());
}

但是这样做,我扫描了三次.一旦建立它,第二次获得最高分,第三次过滤它,这不是最佳的.此外,可能性的数量可能很大(> 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得分(可能性)方法因为可能性null.同样涉及同一行:

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

我不明白怎么list.get(0)可能回到null...

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

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

• 添加synchronizedBestCollector的所有方法: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()
• 每个工作人员将使用该accumulator()功能和自己的本地容器
• combiner()一旦两个工作线程完成工作,将使用它
• 该finisher()时,将使用所有工作线程完成自己的工作和所有的容器已合并

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

您不需要synchronized此处的关键字和并发集合.

顺便提一下,一个Comparator形式的(p1, p2) -> score(p1).compareTo(score(p2))可利用来实现Comparator.comparing(p -> score(p)),或者如果分数值是int:Comparator.comparingInt(p -> score(p)).

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

理解向synchronized关键字accumulator()或combiner()方法添加关键字不能保护通过lambda表达式构造的函数也很重要.它将保护构造函数实例的方法,但不保护函数的代码本身.与内部类相比,没有办法synchronized在实际函数的实现方法中添加关键字.