java流中的异常处理reduce lambda

Question

我正在尝试处理 java 中流的

reduce(item,aggregator)

函数内的异常。 这就是我的原始代码：

List<ReportRow> totalList = newList.stream()
                .collect(Collectors.groupingBy(a -> a.getEngagementCode()))
                .entrySet().stream()
                .map(engagement -> engagement.getValue().stream()
                        .reduce((item, aggregator) -> 
                                new ReportRow(item.getEnvironment(), item.getApplicationName(), item.getEngagementCode(), item.getTotalHits() + aggregator.getTotalHits(), item.getServiceLine(), Float.toString(Float.parseFloat(item.getTotalCost().replace(",", "")) + Float.parseFloat(aggregator.getTotalCost().replace(",", ""))), item.getPrimaryOwnerEmail(), item.getDeploymentId()))
                        .get())
                .collect(Collectors.toList());

这大致就是我期望的样子（虽然不起作用）。

List<ReportRow> totalList = newList.stream()
                .collect(Collectors.groupingBy(a -> a.getEngagementCode()))
                .entrySet().stream()
                .map(engagement -> engagement.getValue().stream()
                        .reduce((item, aggregator) -> {
                            try {
                                new ReportRow(item.getEnvironment(), item.getApplicationName(), item.getEngagementCode(), item.getTotalHits() + aggregator.getTotalHits(), item.getServiceLine(), Float.toString(Float.parseFloat(item.getTotalCost().replace(",", "")) + Float.parseFloat(aggregator.getTotalCost().replace(",", ""))), item.getPrimaryOwnerEmail(), item.getDeploymentId()))
                            } catch (Exception e) {
                                throw new ImproperDataException("No Deployment Id present in row: "+ item.toString());
                            }
                        }
                        .get())
                .collect(Collectors.toList());

当我尝试调用

ReportRow()

构造函数时，会发生异常。 在仍然像原始方法一样使用收集、流和映射的同时处理此异常的方法是什么？

Answer 1

如果发生未经检查的异常，它将终止流处理。这在出现无法恢复的意外问题时非常有用。为此，ImproperDataException 应扩展 RuntimeException。出于可读性原因，我将在方法内移动 try-catch 和 throw 逻辑。

对于受检查的异常，您应该决定需要发生什么。要么继续该过程（通过跳过该项目或使用默认值），要么通过将其包装在未经检查的异常中来终止。您无法从流表达式传播已检查的异常。

这是一个简化且稍显做作的示例。当 ArithmeticException 发生时，报告中将跳过 costPerHit 计算。

static class Value {
    int hits;
    int cost;

    Value(int hits, int cost) {
        this.hits = hits;
        this.cost = cost;
    }
}

static class Report {
    int values;
    int hits;
    int cost;
    double costPerHit;

    Report accumulate(Value value) {
        values++;
        hits += value.hits;
        cost += value.cost;
        updateCostPerHit();
        return this;
    }

    Report combine(Report report) {
        values += report.values;
        hits += report.hits;
        cost += report.cost;
        updateCostPerHit();
        return this;
    }

    private void updateCostPerHit() {
        try {
            costPerHit = cost / hits;
        } catch (ArithmeticException e) {
            e.printStackTrace();
        }
    }

    @Override
    public String toString() {
        return String.format("Report{values=%d, hits=%d, cost=%d, costPerHit=%.2f}", values, hits, cost, costPerHit);
    }
}

public static void main(String[] args) {

    List<Value> values = List.of(
        new Value(0, 3), // causes ArithmeticException
        new Value(5, 1),
        new Value(1, 2),
        new Value(3, 8)
    );

    Report output = values.stream().reduce(new Report(), Report::accumulate, Report::combine);

    System.out.println(output);
}

输出：

java.lang.ArithmeticException: / by zero
    at ErrorHandling$Report.updateCostPerHit(ErrorHandling.java:42)
    at ErrorHandling$Report.accumulate(ErrorHandling.java:28)
    at java.base/java.util.stream.ReduceOps$1ReducingSink.accept(ReduceOps.java:80)
    at java.base/java.util.AbstractList$RandomAccessSpliterator.forEachRemaining(AbstractList.java:720)
    at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
    at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
    at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:913)
    at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
    at java.base/java.util.stream.ReferencePipeline.reduce(ReferencePipeline.java:563)
    at interview.streams.ErrorHandling.main(ErrorHandling.java:63)
Report{values=4, hits=9, cost=14, costPerHit=1.00}

实际上，由于我使用单个 ReportRow 作为某种可变容器，因此最好使用

collect

而不是

reduce

：

Report output = values.stream().collect(Report::new, Report::accumulate, Report::combine);

我认为这也可能更适合您的实际代码