终止或中断Java 8流循环

Question

我有一个包含以下内容的Java 8流循环：

    void matchSellOrder(Market market, Order sellOrder) {
        System.out.println("selling " + market.pair() + " : " + sellOrder);

        market.buyOrders()
                .stream()
                .filter(buyOrder -> buyOrder.price >= sellOrder.price)
                .sorted(BY_ASCENDING_PRICE)
                .forEach((buyOrder) -> {
                    double tradeVolume = Math.min(buyOrder.quantity, sellOrder.quantity);
                    double price = buyOrder.price;

                    buyOrder.quantity -= tradeVolume;
                    sellOrder.quantity -= tradeVolume;

                    Trade trade = new Trade.Builder(market, price, tradeVolume, Trade.Type.SELL).build();
                    CommonUtil.convertToJSON(trade);

                    if (sellOrder.quantity == 0) {
                        System.out.println("order fulfilled");
                        // break loop there
                    }
                });
    }

当满足某个条件时，我该如何跳出循环？

关闭流的正确方法是什么？

更新

我误用了流技术，以为它是一个循环，但实际上它不是为此而设计的。下面是我最终使用的答案提供的代码：

        List<Order> applicableSortedBuyOrders = market.buyOrders()
                .stream()
                .filter(buyOrder -> buyOrder.price >= sellOrder.price)
                .sorted(BY_ASCENDING_PRICE)
                .collect(toList());

        for(Order buyOrder : applicableSortedBuyOrders){
            double tradeVolume = Math.min(buyOrder.quantity, sellOrder.quantity);
            double price = buyOrder.price;

            buyOrder.quantity -= tradeVolume;
            sellOrder.quantity -= tradeVolume;

            Trade trade = new Trade.Builder(market, price, tradeVolume, Trade.Type.SELL).build();
            CommonUtil.printAsJSON(trade);

            if (sellOrder.quantity == 0) {
                System.out.println("order fulfilled");
                break;
            }
        }

- vach

---

这个问题是一个重复问题，但不是指示的那个问题，而是这个问题：https://dev59.com/XmAg5IYBdhLWcg3wrMl5 - Fritz Duchardt

我没有尝试过，但是...如果你在一个peek()消费者中执行你的工作，并在一个并发安全的终止状态对象上跟随一个anyMatch()谓词，会怎样呢？ - simon.watts

你可以在终端操作之前加上 sequential() 以避免并发问题。然后只需将 forEach 替换为 anyMatch 并在谓词内部返回 return sellOrder.quantity == 0。 - benez

Answer 1

Stream.forEach 不是一个循环，也不是为了像 break 这样的方式终止而设计的。如果流是并行流，则 lambda 体可能会同时在不同的线程上执行（难以中断它，并且可能会轻易地产生不正确的结果）。

最好使用迭代器和 while 循环：

Iterator<BuyOrderType> iter = market.buyOrders() // replace BuyOrderType with correct type here
            .stream()
            .filter(buyOrder -> buyOrder.price >= sellOrder.price)
            .sorted(BY_ASCENDING_PRICE).iterator();
while (iter.hasNext()) {
    BuyOrderType buyOrder = iter.next()  // replace BuyOrderType with correct type here
    double tradeVolume = Math.min(buyOrder.quantity, sellOrder.quantity);
    double price = buyOrder.price;

    buyOrder.quantity -= tradeVolume;
    sellOrder.quantity -= tradeVolume;

    Trade trade = new Trade.Builder(market, price, tradeVolume, Trade.Type.SELL).build();
    CommonUtil.convertToJSON(trade);

    if (sellOrder.quantity == 0) {
        System.out.println("order fulfilled");
        break;
    }
}

- fabian

---

谢谢，看来我误用了新功能。 - vach

我同意这个答案，除了第一句话。Stream.forEach是一个循环。不同之处在于它是一个内部循环，而不是外部循环。由于循环是内部的，因此您无法在其上调用break。 - Bogoth

哦，你这点信心都没有！你完全可以跳出 Lambda 循环：`stream.allMatch(value -> { if (...) return true; /* 继续 */ else return false; /* 中断 */ });` - yk4ever

把这个告诉一个并行的流！例如 IntStream.range(0, 1000).boxed().parallel().allMatch(i -> { System.out.println(i); return i % 2 != 0; }); - fabian

@fabian：那就不要使用并行流。 - Matt Cosentino

Answer 2

嗯，就我所知，在流API中没有这样做的方法。

但是如果你确实需要它，可以使用异常。

编辑：对于给予此答案-1的人，我并不是将其作为一个应该遵循的方法进行宣传，它只是一种选项，适用于您需要它的情况，并且回答了问题。

public class BreakException extends RuntimeException {...}

try {
    market.buyOrders()
            .stream()
            .filter(buyOrder -> buyOrder.price >= sellOrder.price)
            .sorted(BY_ASCENDING_PRICE)
            .forEach((buyOrder) -> {
                double tradeVolume = Math.min(buyOrder.quantity, sellOrder.quantity);
                double price = buyOrder.price;

                buyOrder.quantity -= tradeVolume;
                sellOrder.quantity -= tradeVolume;

                Trade trade = new Trade.Builder(market, price, tradeVolume, Trade.Type.SELL).build();
                CommonUtil.convertToJSON(trade);

                if (sellOrder.quantity == 0) {
                    System.out.println("order fulfilled");
                    throw new BreakException()
                }
            });
} catch (BreakException e) {
    //Stoped
}

- Tiago Engel

---

这也是一种选择，但据我所知，抛出和处理异常在性能方面非常耗费资源，应该尽量避免使用（以及上面提到的finally块）。 - vach

不支持并行流。 - rudi

使用异常来控制程序流程不是一个好的实践。 - clapas