public static <E> Stream<E> proxy(Stream<E> src) {
Class<Stream<E>> sClass=(Class)Stream.class;
Class<Spliterator<E>> spClass=(Class)Spliterator.class;
return proxy(src, sClass, spClass, StreamSupport::stream);
}
public static IntStream proxy(IntStream src) {
return proxy(src, IntStream.class, Spliterator.OfInt.class, StreamSupport::intStream);
}
public static LongStream proxy(LongStream src) {
return proxy(src, LongStream.class, Spliterator.OfLong.class, StreamSupport::longStream);
}
public static DoubleStream proxy(DoubleStream src) {
return proxy(src, DoubleStream.class, Spliterator.OfDouble.class, StreamSupport::doubleStream);
}
static final Object EMPTY=new StringBuilder("empty");
static <E,S extends BaseStream<E,S>, Sp extends Spliterator<E>> S proxy(
S src, Class<S> sc, Class<Sp> spc, BiFunction<Sp,Boolean,S> f) {
final class Node<T> implements InvocationHandler,Runnable,
Consumer<Object>, IntConsumer, LongConsumer, DoubleConsumer {
final Class<? extends Spliterator> type;
Spliterator<T> src;
Object first=EMPTY, last=EMPTY;
Node<T> left, right;
Object currConsumer;
public Node(Spliterator<T> src, Class<? extends Spliterator> type) {
this.src = src;
this.type=type;
}
private void value(Object t) {
if(first==EMPTY) first=t;
last=t;
}
public void accept(Object t) {
value(t); ((Consumer)currConsumer).accept(t);
}
public void accept(int t) {
value(t); ((IntConsumer)currConsumer).accept(t);
}
public void accept(long t) {
value(t); ((LongConsumer)currConsumer).accept(t);
}
public void accept(double t) {
value(t); ((DoubleConsumer)currConsumer).accept(t);
}
public void run() {
System.out.println();
finish().forEach(System.out::println);
}
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
Node<T> curr=this; while(curr.right!=null) curr=curr.right;
if(method.getName().equals("tryAdvance")||method.getName().equals("forEachRemaining")) {
curr.currConsumer=args[0];
args[0]=curr;
}
if(method.getName().equals("trySplit")) {
Spliterator s=curr.src.trySplit();
if(s==null) return null;
Node<T> pfx=new Node<>(s, type);
pfx.left=curr.left; curr.left=pfx;
curr.right=new Node<>(curr.src, type);
src=null;
return pfx.create();
}
return method.invoke(curr.src, args);
}
Object create() {
return Proxy.newProxyInstance(null, new Class<?>[]{type}, this);
}
String pad(String s, int left, int len) {
if (len == s.length())
return s;
char[] result = new char[len];
Arrays.fill(result, ' ');
s.getChars(0, s.length(), result, left);
return new String(result);
}
public List<String> finish() {
String cur = toString();
if (left == null) {
return Collections.singletonList(cur);
}
List<String> l = left.finish();
List<String> r = right.finish();
int len1 = l.get(0).length();
int len2 = r.get(0).length();
int totalLen = len1 + len2 + 1;
int leftAdd = 0;
if (cur.length() < totalLen) {
cur = pad(cur, (totalLen - cur.length())/2, totalLen);
} else {
leftAdd = (cur.length() - totalLen)/2;
totalLen = cur.length();
}
List<String> result = new ArrayList<>();
result.add(cur);
char[] dashes = new char[totalLen];
Arrays.fill(dashes, ' ');
Arrays.fill(dashes, len1/2 + leftAdd + 1, len1 + len2/2 + 1
+ leftAdd, '_');
int mid = totalLen/2;
dashes[mid] = '/';
dashes[mid + 1] = '\\';
result.add(new String(dashes));
Arrays.fill(dashes, ' ');
dashes[len1/2 + leftAdd] = '|';
dashes[len1 + len2/2 + 1 + leftAdd] = '|';
result.add(new String(dashes));
int maxSize = Math.max(l.size(), r.size());
for (int i = 0; i < maxSize; i++) {
String lstr = l.size() > i ? l.get(i) : String.format("%"
+ len1 + "s", "");
String rstr = r.size() > i ? r.get(i) : String.format("%"
+ len2 + "s", "");
result.add(pad(lstr + " " + rstr, leftAdd, totalLen));
}
return result;
}
private Object first() {
if(left==null) return first;
Object o=left.first();
if(o==EMPTY) o=right.first();
return o;
}
private Object last() {
if(right==null) return last;
Object o=right.last();
if(o==EMPTY) o=left.last();
return o;
}
public String toString() {
Object o=first(), p=last();
return o==EMPTY? "(empty)": "["+o+(o!=p? ".."+p+']': "]");
}
}
Node<E> n=new Node<>(src.spliterator(), spc);
Sp sp=(Sp)Proxy.newProxyInstance(null, new Class<?>[]{n.type}, n);
return f.apply(sp, true).onClose(n);
}
Permette di avvolgere uno spliterator con un proxy che monitorerà le operazioni di divisione e gli oggetti incontrati. La logica della gestione del blocco è simile a quella di Tagir, infatti, ho copiato le sue routine di stampa dei risultati.
È possibile passare la sorgente del flusso o di un flusso con le stesse operazioni già aggiunte. (In quest'ultimo caso, è necessario applicare .parallel()
il prima possibile allo stream). Come spiegato Tagir, nella maggior parte dei casi, il comportamento divisione dipende dalla sorgente e il parallelismo configurato, quindi, nella maggioranza dei casi, il monitoraggio stati intermedi possono cambiare i valori, ma non i pezzi lavorati:
try(IntStream is=proxy(IntStream.range(0, 100).parallel())) {
is.filter(i -> i/20%2==0)
.mapToObj(ix->"\""+ix+'"')
.forEach(s->{});
}
stamperà
[0..99]
___________________________________/\________________________________
| |
[0..49] [50..99]
_________________/\______________ _________________/\________________
| | | |
[0..24] [25..49] [50..74] [75..99]
________/\_____ ________/\_______ ________/\_______ ________/\_______
| | | | | | | |
[0..11] [12..24] [25..36] [37..49] [50..61] [62..74] [75..86] [87..99]
___/\_ ___/\___ ___/\___ ___/\___ ___/\___ ___/\___ ___/\___ ___/\___
| | | | | | | | | | | | | | | |
[0..5] [6..11] [12..17] [18..24] [25..30] [31..36] [37..42] [43..49] [50..55] [56..61] [62..67] [68..74] [75..80] [81..86] [87..92] [93..99]
mentre
try(Stream<String> s=proxy(IntStream.range(0, 100).parallel().filter(i -> i/20%2==0)
.mapToObj(ix->"\""+ix+'"'))) {
s.forEach(str->{});
}
stamperà
["0".."99"]
___________________________________________/\___________________________________________
| |
["0".."49"] ["50".."99"]
____________________/\______________________ ______________________/\___________________
| | | |
["0".."19"] ["40".."49"] ["50".."59"] ["80".."99"]
____________/\_________ ____________/\______ _______/\___________ ____________/\________
| | | | | | | |
["0".."11"] ["12".."19"] (empty) ["40".."49"] ["50".."59"] (empty) ["80".."86"] ["87".."99"]
_____/\___ _____/\_____ ___/\__ _____/\_____ _____/\_____ ___/\__ _____/\__ _____/\_____
| | | | | | | | | | | | | | | |
["0".."5"] ["6".."11"] ["12".."17"] ["18".."19"] (empty) (empty) ["40".."42"] ["43".."49"] ["50".."55"] ["56".."59"] (empty) (empty) ["80"] ["81".."86"] ["87".."92"] ["93".."99"]
Come possiamo vedere qui, stiamo monitorando il risultato di .filter(…).mapToObj(…)
ma i blocchi sono chiaramente determinati dalla sorgente, probabilmente producendo pezzi vuoti a valle a seconda delle condizioni del filtro.
noti che siamo in grado di combinare il monitoraggio della sorgente con il monitoraggio di Tagir collector:
try(IntStream s=proxy(IntStream.range(0, 100))) {
s.parallel().filter(i -> i/20%2==0)
.boxed().collect(parallelVisualize())
.forEach(System.out::println);
}
questo stampa (si noti che l'uscita collect
viene stampato prima):
[0..99]
________________________________/\_______________________________
| |
[0..49] [50..99]
________________/\______________ _______________/\_______________
| | | |
[0..19] [40..49] [50..59] [80..99]
________/\_____ ________/\______ _______/\_______ ________/\_____
| | | | | | | |
[0..11] [12..19] (empty) [40..49] [50..59] (empty) [80..86] [87..99]
___/\_ ___/\___ ___/\__ ___/\___ ___/\___ ___/\__ ___/\_ ___/\___
| | | | | | | | | | | | | | | |
[0..5] [6..11] [12..17] [18..19] (empty) (empty) [40..42] [43..49] [50..55] [56..59] (empty) (empty) [80] [81..86] [87..92] [93..99]
[0..99]
___________________________________/\________________________________
| |
[0..49] [50..99]
_________________/\______________ _________________/\________________
| | | |
[0..24] [25..49] [50..74] [75..99]
________/\_____ ________/\_______ ________/\_______ ________/\_______
| | | | | | | |
[0..11] [12..24] [25..36] [37..49] [50..61] [62..74] [75..86] [87..99]
___/\_ ___/\___ ___/\___ ___/\___ ___/\___ ___/\___ ___/\___ ___/\___
| | | | | | | | | | | | | | | |
[0..5] [6..11] [12..17] [18..24] [25..30] [31..36] [37..42] [43..49] [50..55] [56..61] [62..67] [68..74] [75..80] [81..86] [87..92] [93..99]
Si può vedere chiaramente come i pezzi della partita di elaborazione, ma dopo il filtraggio, alcuni blocchi hanno meno elementi, alcuni dei quali sono completamente vuoti.
Questo è il luogo di dimostrare, in cui i due modi di monitoraggio possono fare una differenza significativa:
try(DoubleStream is=proxy(DoubleStream.iterate(0, i->i+1)).parallel().limit(100)) {
is.boxed()
.collect(parallelVisualize())
.forEach(System.out::println);
}
[0.0..99.0]
___________________________________________________/\________________________________________________
| |
[0.0..49.0] [50.0..99.0]
_________________________/\______________________ _________________________/\________________________
| | | |
[0.0..24.0] [25.0..49.0] [50.0..74.0] [75.0..99.0]
____________/\_________ ____________/\___________ ____________/\___________ ____________/\___________
| | | | | | | |
[0.0..11.0] [12.0..24.0] [25.0..36.0] [37.0..49.0] [50.0..61.0] [62.0..74.0] [75.0..86.0] [87.0..99.0]
_____/\___ _____/\_____ _____/\_____ _____/\_____ _____/\_____ _____/\_____ _____/\_____ _____/\_____
| | | | | | | | | | | | | | | |
[0.0..5.0] [6.0..11.0] [12.0..17.0] [18.0..24.0] [25.0..30.0] [31.0..36.0] [37.0..42.0] [43.0..49.0] [50.0..55.0] [56.0..61.0] [62.0..67.0] [68.0..74.0] [75.0..80.0] [81.0..86.0] [87.0..92.0] [93.0..99.0]
[0.0..10239.0]
_____________________________/\_____
| |
[0.0..1023.0] [1024.0..10239.0]
____________________/\_______
| |
[1024.0..3071.0] [3072.0..10239.0]
____________/\______
| |
[3072.0..6143.0] [6144.0..10239.0]
___/\_______
| |
[6144.0..10239.0] (empty)
Questo dimostra what Tagir already explained, ruscelli con una dimensione sconosciuta dividere male, e anche il fatto che il limit(…)
offre la possibilità di una buona stima (in realtà, il limite infinito + è teoricamente prevedibile), l'implementazione non ne trae alcun vantaggio.
La sorgente viene suddivisa in blocchi utilizzando una dimensione di batch di 1024
, aumentata di 1024
dopo ogni divisione, creando blocchi fuori dall'intervallo imposto da limit
. Possiamo anche vedere come viene separato un prefisso ogni volta.
Ma quando guardiamo all'uscita split del terminale, possiamo vedere che tra questi pezzi in eccesso sono stati eliminati e che è avvenuta un'altra divisione del primo blocco. Dato che questo chunk è back-end da una matrice intermedia che è stata riempita dall'implementazione predefinita sulla prima divisione, non la notiamo all'origine ma possiamo vedere all'azione del terminale che questa matrice è stata divisa (non sorprendentemente) ben bilanciata .
Quindi abbiamo bisogno di entrambi i modi di monitoraggio per ottenere l'immagine completa qui ...
Davvero un'ottima aggiunta, grazie mille! –