Как правильно уменьшить поток в другой поток

Question

Я поток строк и нулям, как

Stream<String> str1 = Stream.of("A","B","C",null,null,"D",null,"E","F",null,"G",null); 

Я хочу, чтобы уменьшить его в другой поток, в котором любая последовательность не пустая строка сочетал, то есть как

Stream<String> str2 = Stream.of("ABC", "", "D", "EF","G") 

Первый способ, который я нашел - создать коллектор, что во-первых, уменьшить полный входной поток для одного объекта со списком всех объединенных строк, а затем создать новый поток из него:

class Acc1 { 
    final private List<String> data = new ArrayList<>(); 
    final private StringBuilder sb = new StringBuilder(); 

    private void accept(final String s) { 
    if (s != null) 
     sb.append(s); 
    else { 
     data.add(sb.toString()); 
     sb.setLength(0); 
    } 
    } 

    public static Collector<String,Acc1,Stream<String>> collector() { 
    return Collector.of(Acc1::new, Acc1::accept, (a,b)-> a, acc -> acc.data.stream()); 
    } 
} 
... 
Stream<String> str2 = str.collect(Acc1.collector()); 

Но в этом случае перед любым использованием, если str2, даже как str2.findFirst(), поток ввода будет полностью обработан. Это время и память трудоемкой операцией и на бесконечности потока от некоторого генератора он не будет работать на всех

Другой способ - создать внешний объект, который будет держать промежуточное состояние и использовать его в flatMap():

class Acc2 { 
    final private StringBuilder sb = new StringBuilder(); 

    Stream<String> accept(final String s) { 
    if (s != null) { 
     sb.append(s); 
     return Stream.empty(); 
    } else { 
     final String result = sb.toString(); 
     sb.setLength(0); 
     return Stream.of(result); 
    } 
    } 
} 
... 
Acc2 acc = new Acc2(); 
Stream<String> str2 = str1.flatMap(acc::accept); 

В этот случай из str1 будет извлекаться только в элементах, которые действительно доступны через str2.

Но использование внешнего объекта, созданного за пределами обработки потока, выглядит уродливо для меня и, вероятно, может вызвать некоторые побочные эффекты, которые я не вижу сейчас. Также, если str2 будет использоваться позже с parallelStream(), это вызовет непредсказуемый результат.

Есть ли более правильная реализация потоковой редукции без этих недостатков?

Answer 1

Редукция или ее изменяемый вариант, collect, всегда является операцией, которая будет обрабатывать все элементы. Ваша операция может быть реализована с помощью пользовательского Spliterator, например.

public static Stream<String> joinGroups(Stream<String> s) { 
    Spliterator<String> sp=s.spliterator(); 
    return StreamSupport.stream(
     new Spliterators.AbstractSpliterator<String>(sp.estimateSize(), 
     sp.characteristics()&Spliterator.ORDERED | Spliterator.NONNULL) { 
      private StringBuilder sb = new StringBuilder(); 
      private String last; 

      public boolean tryAdvance(Consumer<? super String> action) { 
       if(!sp.tryAdvance(str -> last=str)) 
        return false; 
       while(last!=null) { 
        sb.append(last); 
        if(!sp.tryAdvance(str -> last=str)) break; 
       } 
       action.accept(sb.toString()); 
       sb=new StringBuilder(); 
       return true; 
      } 
     }, false); 
} 

, который производит предполагаемые группы, как вы можете проверить с

joinGroups(Stream.of("A","B","C",null,null,"D",null,"E","F",null,"G",null)) 
    .forEach(System.out::println); 

, но и имеет нужное ленивое поведение, проверяемый с помощью

joinGroups(
    Stream.of("A","B","C",null,null,"D",null,"E","F",null,"G",null) 
      .peek(str -> System.out.println("consumed "+str)) 
).skip(1).filter(s->!s.isEmpty()).findFirst().ifPresent(System.out::println); 

---

После второй мысли, Я пришел к этому чуть более эффективному варианту.Он будет включать в себя StringBuilder только если есть по крайней мере два String s, чтобы присоединиться, в противном случае он будет просто использовать уже существующий единственную String экземпляра или буквальную "" строку для пустых групп:

public static Stream<String> joinGroups(Stream<String> s) { 
    Spliterator<String> sp=s.spliterator(); 
    return StreamSupport.stream(
     new Spliterators.AbstractSpliterator<String>(sp.estimateSize(), 
     sp.characteristics()&Spliterator.ORDERED | Spliterator.NONNULL) { 
      private String next; 

      public boolean tryAdvance(Consumer<? super String> action) { 
       if(!sp.tryAdvance(str -> next=str)) 
        return false; 
       String string=next; 
       if(string==null) string=""; 
       else if(sp.tryAdvance(str -> next=str) && next!=null) { 
        StringBuilder sb=new StringBuilder().append(string); 
        do sb.append(next);while(sp.tryAdvance(str -> next=str) && next!=null); 
        string=sb.toString(); 
       } 
       action.accept(string); 
       return true; 
      } 
     }, false); 
} 

Answer 2

Вы можете использовать filter(Predicate<? super T> predicate). Что-то вроде этого (по ссылке ниже):

roster 
    .stream() 
    .filter(
     p -> p.getGender() == Person.Sex.MALE 
      && p.getAge() >= 18 
      && p.getAge() <= 25) 
    .map(p -> p.getEmailAddress()) 
    .forEach(email -> System.out.println(email)); 

Проверить пункт 9 данного руководства с Oracle на Lambda: https://docs.oracle.com/javase/tutorial/java/javaOO/lambdaexpressions.html

Я надеюсь, что я помог.

Имейте славный день. :)

Answer 3

Это довольно сложно реализовать таких сценариев с использованием стандартного Stream API. В свободной StreamEx библиотеке я расширенный стандартный интерфейс потока с помощью методов, которые позволяют выполнять так называемый «частичное восстановление», которая является именно то, что нужно здесь:

StreamEx<String> str1 = StreamEx.of("A","B","C",null,null,"D",null,"E","F",null,"G",null); 
Stream<String> str2 = str1.collapse((a, b) -> a != null, 
          MoreCollectors.filtering(Objects::nonNull, Collectors.joining())); 
str2.map(x -> '"'+x+'"').forEach(System.out::println); 

Выход:

"ABC" 
"" 
"D" 
"EF" 
"G" 

StreamEx.collapse() метода выполняет частичное сокращение потока с использованием поставляемого коллектора. Первый аргумент - это предикат, который применяется к двум смежным оригинальным элементам и должен возвращать true, если они должны быть уменьшены вместе. Здесь мы просто требуем, чтобы первая из пары не была нулевой ((a, b) -> a != null): это означает, что каждая группа заканчивается null, и здесь начинается новая группа. Теперь нам нужно объединить групповые письма: это можно сделать стандартным сборщиком Collectors.joining(). Однако нам нужно также отфильтровать null. Мы можем сделать это с помощью сборщика MoreCollectors.filtering (на самом деле тот же сборщик будет доступен в Java 9 в классе Collectors).

Эта реализация полностью ленива и довольно удобна для параллельной обработки.