Compactando fluxos usando JDK8 com lambda (java.util.stream.Streams.zip)

149

No JDK 8 com lambda b93, havia uma classe java.util.stream.Streams.zip no b93 que poderia ser usada para compactar fluxos (isso é ilustrado no tutorial Explorando o Java8 Lambdas. Parte 1 de Dhananjay Nene ). Esta função:

Cria um fluxo combinado lento e sequencial cujos elementos são o resultado da combinação dos elementos de dois fluxos.

No entanto, em b98 isso desapareceu. De fato, a Streamsclasse não está acessível em java.util.stream no b98 .

Essa funcionalidade foi movida e, em caso afirmativo, como faço para compactar fluxos de forma concisa usando o b98?

O aplicativo que tenho em mente é nesta implementação em Java do Shen , onde substituí a funcionalidade zip no

  • static <T> boolean every(Collection<T> c1, Collection<T> c2, BiPredicate<T, T> pred)
  • static <T> T find(Collection<T> c1, Collection<T> c2, BiPredicate<T, T> pred)

funciona com código bastante detalhado (que não usa a funcionalidade do b98).

artella
fonte
3
Ah, acabei de descobrir que ele parece ter sido removido completamente: mail.openjdk.java.net/pipermail/lambda-libs-spec-observers/…
artella
"Explorando o Java8 Lambdas. Parte 1" - o novo link para este artigo é blog.dhananjaynene.com/2013/02/exploring-java8-lambdas-part-1
Aleksei Egorov

Respostas:

77

Eu também precisava disso, então peguei o código fonte do b93 e o coloquei em uma classe "util". Eu tive que modificá-lo um pouco para trabalhar com a API atual.

Para referência, aqui está o código de trabalho (corra por seu próprio risco ...):

public static<A, B, C> Stream<C> zip(Stream<? extends A> a,
                                     Stream<? extends B> b,
                                     BiFunction<? super A, ? super B, ? extends C> zipper) {
    Objects.requireNonNull(zipper);
    Spliterator<? extends A> aSpliterator = Objects.requireNonNull(a).spliterator();
    Spliterator<? extends B> bSpliterator = Objects.requireNonNull(b).spliterator();

    // Zipping looses DISTINCT and SORTED characteristics
    int characteristics = aSpliterator.characteristics() & bSpliterator.characteristics() &
            ~(Spliterator.DISTINCT | Spliterator.SORTED);

    long zipSize = ((characteristics & Spliterator.SIZED) != 0)
            ? Math.min(aSpliterator.getExactSizeIfKnown(), bSpliterator.getExactSizeIfKnown())
            : -1;

    Iterator<A> aIterator = Spliterators.iterator(aSpliterator);
    Iterator<B> bIterator = Spliterators.iterator(bSpliterator);
    Iterator<C> cIterator = new Iterator<C>() {
        @Override
        public boolean hasNext() {
            return aIterator.hasNext() && bIterator.hasNext();
        }

        @Override
        public C next() {
            return zipper.apply(aIterator.next(), bIterator.next());
        }
    };

    Spliterator<C> split = Spliterators.spliterator(cIterator, zipSize, characteristics);
    return (a.isParallel() || b.isParallel())
           ? StreamSupport.stream(split, true)
           : StreamSupport.stream(split, false);
}
siki
fonte
1
O fluxo resultante não deveria ser SIZEDse um deles é SIZED, não os dois?
Didier L
5
Acho que não. Ambos os fluxos devem ser SIZEDpara que essa implementação funcione. Na verdade, depende de como você define o zíper. Você deve poder compactar dois fluxos de tamanho diferente, por exemplo? Como seria o fluxo resultante então? Acredito que é por isso que essa função foi realmente omitida da API. Há muitas maneiras de fazer isso e cabe ao usuário decidir qual comportamento deve ser o "correto". Você descartaria os elementos do fluxo mais longo ou preencheria a lista mais curta? Se sim, com que valor (es)?
siki
A menos que eu esteja perdendo alguma coisa, não há necessidade de elenco (por exemplo, para Spliterator<A>).
Jb0bs
Existe um site em que o código-fonte do Java 8 b93 esteja hospedado? Estou tendo problemas para encontrá-lo.
Starwarswii 07/07
42

zip é uma das funções fornecidas pela biblioteca protonpack .

Stream<String> streamA = Stream.of("A", "B", "C");
Stream<String> streamB  = Stream.of("Apple", "Banana", "Carrot", "Doughnut");

List<String> zipped = StreamUtils.zip(streamA,
                                      streamB,
                                      (a, b) -> a + " is for " + b)
                                 .collect(Collectors.toList());

assertThat(zipped,
           contains("A is for Apple", "B is for Banana", "C is for Carrot"));
Dominic Fox
fonte
1
Também encontrado em StreamEx: amaembo.github.io/streamex/javadoc/one/util/streamex/...
tokland
34

Se você possui o Guava em seu projeto, pode usar o método Streams.zip (adicionado no Guava 21):

Retorna um fluxo no qual cada elemento é o resultado da passagem do elemento correspondente de cada fluxoA e fluxoB para a função. O fluxo resultante será apenas o menor dos dois fluxos de entrada; se um fluxo for mais longo, seus elementos extras serão ignorados. O fluxo resultante não é eficientemente dividido. Isso pode prejudicar o desempenho paralelo.

 public class Streams {
     ...

     public static <A, B, R> Stream<R> zip(Stream<A> streamA,
             Stream<B> streamB, BiFunction<? super A, ? super B, R> function) {
         ...
     }
 }
ZhekaKozlov
fonte
26

Compactando dois fluxos usando JDK8 com lambda ( gist ).

public static <A, B, C> Stream<C> zip(Stream<A> streamA, Stream<B> streamB, BiFunction<A, B, C> zipper) {
    final Iterator<A> iteratorA = streamA.iterator();
    final Iterator<B> iteratorB = streamB.iterator();
    final Iterator<C> iteratorC = new Iterator<C>() {
        @Override
        public boolean hasNext() {
            return iteratorA.hasNext() && iteratorB.hasNext();
        }

        @Override
        public C next() {
            return zipper.apply(iteratorA.next(), iteratorB.next());
        }
    };
    final boolean parallel = streamA.isParallel() || streamB.isParallel();
    return iteratorToFiniteStream(iteratorC, parallel);
}

public static <T> Stream<T> iteratorToFiniteStream(Iterator<T> iterator, boolean parallel) {
    final Iterable<T> iterable = () -> iterator;
    return StreamSupport.stream(iterable.spliterator(), parallel);
}
Karol Król
fonte
2
Solução agradável e (relativamente) compacta! Requer que você coloque import java.util.function.*;e import java.util.stream.*;na parte superior do seu arquivo.
sffc
Observe que esta é uma operação do terminal no fluxo. Isto significa que para infinitas fluxos, este método divide
smac89
2
Tanto invólucros inúteis: Aqui () -> iteratore aqui novamente: iterable.spliterator(). Por que não implementar diretamente um em Spliteratorvez de um Iterator? Verificar @Doradus answer stackoverflow.com/a/46230233/1140754
Miguel Gamboa
20

Como não consigo conceber nenhum uso de zipar em coleções que não sejam indexadas (Listas) e sou um grande fã de simplicidade, esta seria minha solução:

<A,B,C>  Stream<C> zipped(List<A> lista, List<B> listb, BiFunction<A,B,C> zipper){
     int shortestLength = Math.min(lista.size(),listb.size());
     return IntStream.range(0,shortestLength).mapToObj( i -> {
          return zipper.apply(lista.get(i), listb.get(i));
     });        
}
Rafael
fonte
1
Eu acho que mapToObjectdeveria ser mapToObj.
Seanf 03/03/19
se a lista não estiver RandomAccess(por exemplo, nas listas vinculadas), isso será muito lento
avmohan
Definitivamente. Mas a maioria dos desenvolvedores de Java está ciente de que o LinkedList tem um desempenho ruim para operações de acesso ao índice.
Rafael
11

Os métodos da classe que você mencionou foram movidos para a Streamprópria interface em favor dos métodos padrão. Mas parece que o zipmétodo foi removido. Talvez porque não esteja claro qual deve ser o comportamento padrão para fluxos de tamanhos diferentes. Mas implementar o comportamento desejado é direto:

static <T> boolean every(
  Collection<T> c1, Collection<T> c2, BiPredicate<T, T> pred) {
    Iterator<T> it=c2.iterator();
    return c1.stream().allMatch(x->!it.hasNext()||pred.test(x, it.next()));
}
static <T> T find(Collection<T> c1, Collection<T> c2, BiPredicate<T, T> pred) {
    Iterator<T> it=c2.iterator();
    return c1.stream().filter(x->it.hasNext()&&pred.test(x, it.next()))
      .findFirst().orElse(null);
}
Holger
fonte
Você não predicatepassou para o filtro com estado ? Isso viola o contrato do método e, especialmente, não funciona ao processar o fluxo em paralelo.
Andreas
2
@ Andreas: nenhuma das soluções aqui suporta processamento paralelo. Como meus métodos não retornam um fluxo, eles garantem que os fluxos não sejam executados em paralelo. Da mesma forma, o código da resposta aceita retorna um fluxo que pode ser transformado em paralelo, mas na verdade não fará nada em paralelo. Dito isto, os predicados estaduais são desencorajados, mas não violam o contrato. Eles podem até ser usados ​​em contexto paralelo se você garantir que a atualização de estado seja segura para threads. Em algumas situações, eles são inevitáveis, por exemplo, transformar um fluxo em distinto é um predicado estatal em si .
21915 Holger
2
@ Andreas: você pode adivinhar por que essas operações foram removidas da API Java ...
Holger
8

Eu humildemente sugiro esta implementação. O fluxo resultante é truncado para o menor dos dois fluxos de entrada.

public static <L, R, T> Stream<T> zip(Stream<L> leftStream, Stream<R> rightStream, BiFunction<L, R, T> combiner) {
    Spliterator<L> lefts = leftStream.spliterator();
    Spliterator<R> rights = rightStream.spliterator();
    return StreamSupport.stream(new AbstractSpliterator<T>(Long.min(lefts.estimateSize(), rights.estimateSize()), lefts.characteristics() & rights.characteristics()) {
        @Override
        public boolean tryAdvance(Consumer<? super T> action) {
            return lefts.tryAdvance(left->rights.tryAdvance(right->action.accept(combiner.apply(left, right))));
        }
    }, leftStream.isParallel() || rightStream.isParallel());
}
Doradus
fonte
Eu gosto da sua proposta. Mas não concordo totalmente com o último .., leftStream.isParallel() || rightStream.isParallel(). Eu acho que não tem efeito, porque AbstractSpliteratoroferece paralelismo limitado por padrão. Então eu acho que o resultado final será o mesmo que passar false.
Miguel Gamboa
@MiguelGamboa - obrigado pelo seu comentário. Não sei ao certo o que você quer dizer com "paralelismo limitado por padrão" - você tem um link para alguns documentos?
Doradus
6

A biblioteca Lazy-Seq fornece funcionalidade zip.

https://github.com/nurkiewicz/LazySeq

Esta biblioteca é fortemente inspirada scala.collection.immutable.Streame visa fornecer uma implementação de sequência lenta, imutável, segura para threads e fácil de usar, possivelmente infinita.

Nick Siderakis
fonte
5

Usando a biblioteca Guava mais recente (para a Streamsclasse), você poderá fazer

final Map<String, String> result = 
    Streams.zip(
        collection1.stream(), 
        collection2.stream(), 
        AbstractMap.SimpleEntry::new)
    .collect(Collectors.toMap(e -> e.getKey(), e  -> e.getValue()));
Dan Borza
fonte
2

Isso funcionaria para você? É uma função curta, que avalia preguiçosamente os fluxos que estão sendo compactados, para que você possa fornecê-los com fluxos infinitos (não é necessário ter o tamanho dos fluxos sendo compactados).

Se os fluxos são finitos, eles param assim que um dos fluxos fica sem elementos.

import java.util.Objects;
import java.util.function.BiFunction;
import java.util.stream.Stream;

class StreamUtils {
    static <ARG1, ARG2, RESULT> Stream<RESULT> zip(
            Stream<ARG1> s1,
            Stream<ARG2> s2,
            BiFunction<ARG1, ARG2, RESULT> combiner) {
        final var i2 = s2.iterator();
        return s1.map(x1 -> i2.hasNext() ? combiner.apply(x1, i2.next()) : null)
                .takeWhile(Objects::nonNull);
    }
}

Aqui está um código de teste de unidade (muito mais longo que o próprio código!)

import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static org.junit.jupiter.api.Assertions.assertEquals;

class StreamUtilsTest {
    @ParameterizedTest
    @MethodSource("shouldZipTestCases")
    <ARG1, ARG2, RESULT>
    void shouldZip(
            String testName,
            Stream<ARG1> s1,
            Stream<ARG2> s2,
            BiFunction<ARG1, ARG2, RESULT> combiner,
            Stream<RESULT> expected) {
        var actual = StreamUtils.zip(s1, s2, combiner);

        assertEquals(
                expected.collect(Collectors.toList()),
                actual.collect(Collectors.toList()),
                testName);
    }

    private static Stream<Arguments> shouldZipTestCases() {
        return Stream.of(
                Arguments.of(
                        "Two empty streams",
                        Stream.empty(),
                        Stream.empty(),
                        (BiFunction<Object, Object, Object>) StreamUtilsTest::combine,
                        Stream.empty()),
                Arguments.of(
                        "One singleton and one empty stream",
                        Stream.of(1),
                        Stream.empty(),
                        (BiFunction<Object, Object, Object>) StreamUtilsTest::combine,
                        Stream.empty()),
                Arguments.of(
                        "One empty and one singleton stream",
                        Stream.empty(),
                        Stream.of(1),
                        (BiFunction<Object, Object, Object>) StreamUtilsTest::combine,
                        Stream.empty()),
                Arguments.of(
                        "Two singleton streams",
                        Stream.of("blah"),
                        Stream.of(1),
                        (BiFunction<Object, Object, Object>) StreamUtilsTest::combine,
                        Stream.of(pair("blah", 1))),
                Arguments.of(
                        "One singleton, one multiple stream",
                        Stream.of("blob"),
                        Stream.of(2, 3),
                        (BiFunction<Object, Object, Object>) StreamUtilsTest::combine,
                        Stream.of(pair("blob", 2))),
                Arguments.of(
                        "One multiple, one singleton stream",
                        Stream.of("foo", "bar"),
                        Stream.of(4),
                        (BiFunction<Object, Object, Object>) StreamUtilsTest::combine,
                        Stream.of(pair("foo", 4))),
                Arguments.of(
                        "Two multiple streams",
                        Stream.of("nine", "eleven"),
                        Stream.of(10, 12),
                        (BiFunction<Object, Object, Object>) StreamUtilsTest::combine,
                        Stream.of(pair("nine", 10), pair("eleven", 12)))
        );
    }

    private static List<Object> pair(Object o1, Object o2) {
        return List.of(o1, o2);
    }

    static private <T1, T2> List<Object> combine(T1 o1, T2 o2) {
        return List.of(o1, o2);
    }

    @Test
    void shouldLazilyEvaluateInZip() {
        final var a = new AtomicInteger();
        final var b = new AtomicInteger();
        final var zipped = StreamUtils.zip(
                Stream.generate(a::incrementAndGet),
                Stream.generate(b::decrementAndGet),
                (xa, xb) -> xb + 3 * xa);

        assertEquals(0, a.get(), "Should not have evaluated a at start");
        assertEquals(0, b.get(), "Should not have evaluated b at start");

        final var takeTwo = zipped.limit(2);

        assertEquals(0, a.get(), "Should not have evaluated a at take");
        assertEquals(0, b.get(), "Should not have evaluated b at take");

        final var list = takeTwo.collect(Collectors.toList());

        assertEquals(2, a.get(), "Should have evaluated a after collect");
        assertEquals(-2, b.get(), "Should have evaluated b after collect");
        assertEquals(List.of(2, 4), list);
    }
}
dominic
fonte
Eu tive que largar o takeWhileno final era que não parece estar em java8, mas não é um problema, pois o receptor pode filtrar quaisquer valores nulos que ocorrem quando os fluxos compactados não são do mesmo tamanho. Eu acho que essa resposta deve ser a resposta número 1, pois é consistente e compreensível. ótimo trabalho obrigado novamente.
simbo1905
1
public class Tuple<S,T> {
    private final S object1;
    private final T object2;

    public Tuple(S object1, T object2) {
        this.object1 = object1;
        this.object2 = object2;
    }

    public S getObject1() {
        return object1;
    }

    public T getObject2() {
        return object2;
    }
}


public class StreamUtils {

    private StreamUtils() {
    }

    public static <T> Stream<Tuple<Integer,T>> zipWithIndex(Stream<T> stream) {
        Stream<Integer> integerStream = IntStream.range(0, Integer.MAX_VALUE).boxed();
        Iterator<Integer> integerIterator = integerStream.iterator();
        return stream.map(x -> new Tuple<>(integerIterator.next(), x));
    }
}
robby_pelssers
fonte
1

O cyclops-react da AOL , com o qual contribuo, também fornece a funcionalidade de compactar, tanto por meio de uma implementação estendida do Stream , que também implementa a interface do reative-streams ReactiveSeq quanto pelo StreamUtils, que oferece grande parte da mesma funcionalidade por métodos estáticos aos Java Streams padrão.

 List<Tuple2<Integer,Integer>> list =  ReactiveSeq.of(1,2,3,4,5,6)
                                                  .zip(Stream.of(100,200,300,400));


  List<Tuple2<Integer,Integer>> list = StreamUtils.zip(Stream.of(1,2,3,4,5,6),
                                                  Stream.of(100,200,300,400));

Ele também oferece um zíper baseado em aplicativo mais generalizado. Por exemplo

   ReactiveSeq.of("a","b","c")
              .ap3(this::concat)
              .ap(of("1","2","3"))
              .ap(of(".","?","!"))
              .toList();

   //List("a1.","b2?","c3!");

   private String concat(String a, String b, String c){
    return a+b+c;
   }

E até a capacidade de emparelhar todos os itens em um fluxo com todos os itens em outro

   ReactiveSeq.of("a","b","c")
              .forEach2(str->Stream.of(str+"!","2"), a->b->a+"_"+b);

   //ReactiveSeq("a_a!","a_2","b_b!","b_2","c_c!","c2")
John McClean
fonte
0

Se alguém precisar disso ainda, há uma StreamEx.zipWithfunção na biblioteca streamex :

StreamEx<String> givenNames = StreamEx.of("Leo", "Fyodor")
StreamEx<String> familyNames = StreamEx.of("Tolstoy", "Dostoevsky")
StreamEx<String> fullNames = givenNames.zipWith(familyNames, (gn, fn) -> gn + " " + fn);

fullNames.forEach(System.out::println);  // prints: "Leo Tolstoy\nFyodor Dostoevsky\n"
const.grigoryev
fonte
-1

Isso é ótimo. Eu tive que compactar dois fluxos em um mapa, com um fluxo sendo a chave e outro sendo o valor

Stream<String> streamA = Stream.of("A", "B", "C");
Stream<String> streamB  = Stream.of("Apple", "Banana", "Carrot", "Doughnut");    
final Stream<Map.Entry<String, String>> s = StreamUtils.zip(streamA,
                    streamB,
                    (a, b) -> {
                        final Map.Entry<String, String> entry = new AbstractMap.SimpleEntry<String, String>(a, b);
                        return entry;
                    });

System.out.println(s.collect(Collectors.toMap(e -> e.getKey(), e -> e.getValue())));

Saída: {A = Maçã, B = Banana, C = Cenoura}

Gnana
fonte