Как выполнить операцию сокращения на некоторых участках потока

Question

Я столкнулся с ситуацией, которую я хотя бы обрабатывал с помощью Stream API, но я просто не могу найти правильное решение.

Дело в следующем: у меня есть поток элементов, отсортированных по полю идентификатора. Для этого идентификатора существует несколько элементов с одинаковым значением, и мне нужно дедуплицировать их на основе условий в других полях. Концептуально это можно рассматривать как операцию уменьшения на нескольких кусках потока, приводя к потоку того же типа.

На данный момент единственным решением, с которым я справляюсь, является сбор потока на основе общего идентификатора, чтобы получить что-то вроде Map<Id, List<Elem>>, а затем использовать поток этой карты для применения моих правил дедупликации и продолжения. Проблема (и почему я не буду использовать это решение) заключается в том, что collect - это операция терминала, повторная потоковая передача после нее означает, что я буду перебирать элементы по два раза.

UPDATE

Рассмотрим следующий класс:

public static class Item { 
    private final int _id; 
    private final double _price; 

    public Item(final int id, final double price) { 
     _id = id; 
     _price = price; 
    } 

    public int id() { 
     return _id; 
    } 

    public double price() { 
     return _price; 
    } 
} 

И следующий поток:

final Stream<Item> items = Stream.<Item>builder() 
     .add(new Item(1, 4)) 
     .add(new Item(1, 6)) 
     .add(new Item(1, 3)) 
     .add(new Item(2, 5)) 
     .add(new Item(2, 1)) 
     .add(new Item(3, 5)) 
     .build(); 

После требуемой операции, если правило дедупликации является «с самым высоким цена ", поток должен содержать только Item (1, 6), Item (2, 5) и Item (3, 5).

Если я делаю это настоятельно, я могу использовать свои предметы, пока они имеют один и тот же идентификатор, резервируя их во временной коллекции и дедуплицируя эту коллекцию, когда сталкиваются с элементом с другим идентификатором.

Если я использую коллекцию для первой группы элементов по идентификатору, я буду использовать все данные одновременно, прежде чем переходить к следующей операции, и мне нужно избегать этого.

Answer 1

Для большинства случаев такого рода временное хранение, например, Map, неизбежно. В конце концов, это эффективный алгоритм поиска карты, который позволяет идентифицировать группу, к которой принадлежит каждый элемент. Кроме того, возможно, что первая группа содержит первый и последний элемент исходного потока и единственный способ выяснить, является ли это так, итерирует весь поток источника. Это может быть неверно для вашего специального случая предварительно отсортированных данных, но API не дает возможности использовать его для операции группировки. И он не играл бы хорошо с поддержкой параллельного потока, если бы он существовал.

Но рассмотрите groupingBy collector accepting a downstream Collector, который позволяет свести группы к их окончательному результату на месте. Если это истинное сокращение, вы можете использовать, например. reducing как нисходящий коллектор. Это позволяет собирать элементы в Map<Id, Reduced>, а не Map<Id, List<Elem>>, поэтому вы не собираете в List с, которые впоследствии должны быть уменьшены.

Для любого подобного случая, если вы можете описать последующую операцию как Collector, ее обработка действительно начнется сразу же при встрече с первым элементом группы. Обратите внимание, что есть и другие комбинации Collector s, такие как mapping и collectingAndThen. Java 9 также добавит filtering и flatMapping, так что вы можете выразить много типичных операций Stream в виде нисходящего коллектора. Для удобства this collector сочетает в себе этап отображения с последующим этапом сокращения.

Дальнейшая обработка групп может быть выполнена только после полного завершения группировки, путем доступа к Map.values(). Если конечный результат должен быть Collection, нет необходимости перетекать через него снова, поскольку существующие операции сбора достаточны, например. вы можете использовать new ArrayList<>(map.values()), если вам нужен List, а не неспецифический Collection.

---

Если ваша проблема в том, что эта операция не должна выполняться до тех пор, пока вызывающий не начинает терминальный операцию на конечном потоке, вы можете использовать операцию так:

public Stream<ResultType> stream() { 
    return StreamSupport.stream(() -> items.stream() 
      .collect(Collectors.groupingBy(classificationFunc, 
       Collectors.reducing(id, mappingFunc, reductionFunc))) 
      .values().spliterator(), 
     Spliterator.SIZED, false); 
} 

Answer 2

Я не проверял , но используя StreamEx библиотеку, вы должны быть в состоянии collapse() соседних элементов, как это:

items.collapse((a, b) -> a.id() == b.id(), (a, b) -> a.price() < b.price() ? b : a)