Clojure dönüştürücüleri, Java'daki akışlarda ara işlemlerle aynı kavram mıdır?

Question

Clojure'daki transdüserler hakkında bilgi sahibi olduğum için bana anımsattığı şeyi anımsattı: Java 8 akışları!

Transducers are composable algorithmic transformations. They are independent from the context of their input and output sources and specify only the essence of the transformation in terms of an individual element.

A stream is not a data structure that stores elements; instead, it conveys elements from a source such as a data structure, an array, a generator function, or an I/O channel, through a pipeline of computational operations.

Clojure:

(def xf 
    (comp 
    (filter odd?) 
    (map inc) 
    (take 5))) 

(println 
    (transduce xf + (range 100))) ; => 30 
(println 
    (into [] xf (range 100)))  ; => [2 4 6 8 10] 

Java:

// Purposely using Function and boxed primitive streams (instead of 
// UnaryOperator<LongStream>) in order to keep it general. 
Function<Stream<Long>, Stream<Long>> xf = 
     s -> s.filter(n -> n % 2L == 1L) 
       .map(n -> n + 1L) 
       .limit(5L); 

System.out.println(
     xf.apply(LongStream.range(0L, 100L).boxed()) 
       .reduce(0L, Math::addExact)); // => 30 
System.out.println(
     xf.apply(LongStream.range(0L, 100L).boxed()) 
       .collect(Collectors.toList())); // => [2, 4, 6, 8, 10] 

Statik/dinamik yazımdaki farklılıktan ayrı olarak, bunlar bana kullanım ve kullanım açısından oldukça benzer gözüküyor.

Java akışlarının dönüştürmeleri ile benzetmenin dönüştürücüler hakkında mantıklı bir yol olduğu mu? Eğer değilse, nasıl kusurludur, ya da ikisi kavram açısından nasıl farklıdır (uygulamadan söz edilmez)?

Answer 1

temel fark fiiller (operasyonlar) kümesi nedense bu dönüştürücüler için açık iken akışları için kapalı olduğunu geçerli:

import java.util.function.Function; 
import java.util.function.Supplier; 
import java.util.stream.Stream; 
import java.util.stream.Stream.Builder; 

public class StreamUtils { 
    static <T> Stream<T> delay(final Supplier<Stream<T>> thunk) { 
     return Stream.of((Object) null).flatMap(x -> thunk.get()); 
    } 

    static class Partitioner<T> implements Function<T, Stream<Stream<T>>> { 
     final Function<T, ?> f; 

     Object prev; 
     Builder<T> sb; 

     public Partitioner(Function<T, ?> f) { 
      this.f = f; 
     } 

     public Stream<Stream<T>> apply(T t) { 
      Object tag = f.apply(t); 
      if (sb != null && prev.equals(tag)) { 
       sb.accept(t); 
       return Stream.empty(); 
      } 
      Stream<Stream<T>> partition = sb == null ? Stream.empty() : Stream.of(sb.build()); 
      sb = Stream.builder(); 
      sb.accept(t); 
      prev = tag; 
      return partition; 
     } 

     Stream<Stream<T>> flush() { 
      return sb == null ? Stream.empty() : Stream.of(sb.build()); 
     } 
    } 

    static <T> Stream<Stream<T>> partitionBy(Stream<T> in, Function<T, ?> f) { 
     Partitioner<T> partitioner = new Partitioner<>(f); 
     return Stream.concat(in.flatMap(partitioner), delay(() -> partitioner.flush())); 
    } 
} 

: akışlarında partition uygulamaya örneğin denemek, biraz ikinci sınıf hissediyor Ayrıca diziler ve redüktörler gibi, dönüştüğünüzde "daha büyük" bir hesaplama oluşturmazsanız, "daha büyük" bir kaynak yaratırsınız.

Hesaplamaları geçebilmek için, yöntemlerden birinci sınıf varlıklara (kaynaklardan çözülecek şekilde) işlemleri kaldırmak için Akıştan Akışa bir işlevi xf tanıttınız. Bunu yaparak çok büyük bir arayüzle de olsa bir dönüştürücü oluşturdunuz. benim eğitimsiz gözlerle için

import java.util.function.Function; 
import java.util.function.Supplier; 
import java.util.stream.Stream; 
import java.util.stream.Stream.Builder; 

import clojure.lang.AFn; 
import clojure.lang.IFn; 
import clojure.lang.Reduced; 

public class StreamUtils { 
    static <T> Stream<T> delay(final Supplier<Stream<T>> thunk) { 
     return Stream.of((Object) null).flatMap(x -> thunk.get()); 
    } 

    static class Transducer implements Function { 
     IFn rf; 

     public Transducer(IFn xf) { 
      rf = (IFn) xf.invoke(new AFn() { 
       public Object invoke(Object acc) { 
        return acc; 
       } 

       public Object invoke(Object acc, Object item) { 
        ((Builder<Object>) acc).accept(item); 
        return acc; 
       } 
      }); 
     } 

     public Stream<?> apply(Object t) { 
      if (rf == null) return Stream.empty(); 
      Object ret = rf.invoke(Stream.builder(), t); 
      if (ret instanceof Reduced) { 
       Reduced red = (Reduced) ret; 
       Builder<?> sb = (Builder<?>) red.deref(); 
       return Stream.concat(sb.build(), flush()); 
      } 
      return ((Builder<?>) ret).build(); 
     } 

     Stream<?> flush() { 
      if (rf == null) return Stream.empty(); 
      Builder<?> sb = (Builder<?>) rf.invoke(Stream.builder()); 
      rf = null; 
      return sb.build(); 
     } 
    } 

    static <T> Stream<?> withTransducer(Stream<T> in, IFn xf) { 
     Transducer transducer = new Transducer(xf); 
     return Stream.concat(in.flatMap(transducer), delay(() -> transducer.flush())); 
    } 
}