标签: java-stream

我想要一个类似的东西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 …

我想计算一个流的不同元素,我想知道为什么

Stream<String> stream = Stream.of("a", "b", "a", "c", "c", "a", "a", "d");
Map<String, Integer> counter1 = stream.collect(Collectors.toMap(s -> s, 1, Integer::sum));

不起作用.Eclipse告诉我

收集器类型中的Map(Function,Function,BinaryOperator)方法不适用于参数((s) - > {},int,Integer :: sum)

顺便说一句,我知道这个解决方案:

Map<String, Long> counter2 = stream.collect(Collectors.groupingBy(s -> s, Collectors.counting()));

所以我有两个问题:

1. 我的第一种方法有什么错误？
2. 你会如何实施这样的计数器？

编辑:我自己解决了第一个问题:

Map<String, Integer> counter1 = stream.collect(Collectors.toMap(s -> s, s -> 1, Integer::sum)); 

Java期望一个函数作为第二个参数.

Java API文档声明combiner该collect方法的参数必须是:

用于组合两个值的关联,非干扰,无状态函数,它必须与累加器函数兼容

A combiner是BiConsumer<R,R>接收两个类型的参数R并返回void.但是文档没有说明我们是否应该将元素组合到第一个或第二个参数中？

例如,以下示例可以给出不同的结果,取决于组合的顺序是: m1.addAll(m2)或m2.addAll(m1).

List<String> res = LongStream
     .rangeClosed(1, 1_000_000)
     .parallel()
     .mapToObj(n -> "" + n)
     .collect(ArrayList::new, ArrayList::add,(m1, m2) -> m1.addAll(m2));

我知道在这种情况下我们可以简单地使用方法句柄,例如ArrayList::addAll.然而,在某些情况下需要Lambda并且我们必须以正确的顺序组合项目,否则我们可能在并行处理时得到不一致的结果.

这是否在Java 8 API文档的任何部分中声明了？或者它真的没关系？

我有一个对象列表A,A有一个名为Address的属性,它有一个街道名称 - streetName

从AI想要获取所有街道名称列表的对象列表中.一个级别的集合似乎在流中非常可行但是如何使用一行代码获得嵌套的String.

因此,从对象AI获取地址列表可以这样做:

listOfObjectsA.stream().map(a::getAddress).collect(Collectors.toList());

我的最终目标是获取街道名称列表,因此我无法使用lambdas找出二级集合.

我无法找到我想要的精确示例.有人可以帮我这个.

我已经为Java 8编写了一个自定义收集器.它的聚合器是一个包含一对列表的Map:

    @Override
    public Supplier<Map<Boolean, List<Object>>> supplier() {
        return () -> {
            Map<Boolean, List<Object>> map = new HashMap<>(2);
            map.put(false, new ArrayList<>());
            map.put(true, new ArrayList<>());
            return map;
        };
    }

所以我认为它的组合是这样的:

    @Override
    public BinaryOperator<Map<Boolean, List<Object>>> combiner() {
        return (a, b) -> {
            a.get(false).addAll(b.get(false));
            a.get(true).addAll(b.get(true));
            return a;
        };
    }

我想测试收集器以确保它是否以及何时并行处理流,结果是正确的.

我怎样才能编写一个单元测试呢？

当然我可以写一个combiner直接调用的测试,但这不是我想要的.我希望证据在收集的背景下起作用.

Javadoc Collector说:

为确保顺序和并行执行产生相同的结果,收集器函数必须满足标识和关联约束.

我可以通过测试这些约束来获得对收集器的信心吗？怎么样？

在java.util.stream.Stream接口中,

<R> R collect(Supplier<R> supplier,
              BiConsumer<R, ? super T> accumulator,
              BiConsumer<R, R> combiner);

组合器是a BiConsumer<R, R>,而在

<R, A> R collect(Collector<? super T, A, R> collector);

组合器是一个BinaryOperator<A>只是一个BiFunction<A,A,A>.

虽然后面的形式清楚地定义了组合后组合对象的参考,但前一种形式却没有.

那么任何Stream实现库如何知道,前一种情况下的组合对象是什么？

可以说我下面有一个类有getter和setter但只有默认构造函数.

注意:我不允许更改此类的结构.

class Target {

    private String year;
    private String month;
    private String name;
    private double target;
    private double achieved;


    public String getYear() {
        return year;
    }

    public void setYear(String year) {
        this.year = year;
    }

    public String getMonth() {
        return month;
    }

    public void setMonth(String month) {
        this.month = month;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public double getTarget() {
        return target;
    }

    public void setTarget(double target) { …

我必须迭代130多个数据传输对象，并且每次都会生成一个json以上传到AWS S3。

如果不进行任何改进，则整个过程大约需要90秒。我尝试使用lamba而不是lamba，两者的结果相同。

for(AbstractDTO dto: dtos) {
    try {
        processDTO(dealerCode, yearPeriod, monthPeriod, dto);
    } catch (FileAlreadyExistsInS3Exception e) {
        failedToUploadDTOs.add(e.getLocalizedMessage() + ": " + dto.fileName() + ".json");
    }
}

dtos.stream().forEach(dto -> {
    try {
        processDTO(dealerCode, yearPeriod, monthPeriod, dto);
    } catch (FileAlreadyExistsInS3Exception e) {
        failedToUploadDTOs.add(e.getLocalizedMessage() + ": " + dto.fileName() + ".json");
    }
});

经过一番调查，我得出结论，processDTO方法运行每项大约需要0.650ms。

我的第一个尝试是使用并行流，结果非常好，大约需要15秒才能完成整个过程：

dtos.parallelStream().forEach(dto -> {
    try {
        processDTO(dealerCode, yearPeriod, monthPeriod, dto);
    } catch (FileAlreadyExistsInS3Exception e) {
        failedToUploadDTOs.add(e.getLocalizedMessage() + ": " …

我有2个对象的数组的列表：

List<Object[2]>

其中object [0]是一个整数，而object [1]是一个字符串。

我如何流式列表并在每个对象上应用不同的功能？这样，结果将是一个具有以下内容的数组：

result[0] = multiplication of all object[0]
result[1] = concatenation of all object[1]

我如何检查流实例是否已被使用（意味着已经调用了终端操作，使得对终端操作的任何进一步调用都可能失败IllegalStateException: stream has already been operated upon or closed.？

理想情况下，我想要一个不消耗流的方法，如果该流尚未被消耗，则返回一个布尔值false（如果流被消耗而没有IllegalStateException从流方法中捕获）（因为对控制流使用Exception代价很高并且容易出错） ，尤其是在使用标准例外时）。

与hasNext()Iterator中类似的方法，它具有异常抛出和布尔返回行为（尽管没有约定to next()）。

例：

public void consume(java.util.function.Consumer<Stream<?>> consumer, Stream<?> stream) {
   consumer.accept(stream);
   // defensive programming, check state
   if (...) {
       throw new IllegalStateException("consumer must call terminal operation on stream");
   }
}

目标是如果客户端代码在不消耗流的情况下调用此方法，则会尽早失败。

似乎没有办法做到这一点，我必须添加一个try-catch块来调用任何终端操作，例如iterator()，捕获异常并抛出一个新异常。

可以接受的答案也可以是“不存在解决方案”，其中有充分的理由说明为什么规范无法添加这种方法（如果存在良好的理由）。似乎JDK流通常在其终端方法的开头具有以下代码片段：

// in AbstractPipeline.java
if (linkedOrConsumed)
    throw new IllegalStateException(MSG_STREAM_LINKED);

因此，对于那些流，实现这种方法似乎并不那么困难。