Возможно ли иметь Java Stream с промежуточными производителями? - PullRequest
Новая
       12

Возможно ли иметь Java Stream с промежуточными производителями?

0 голосов
/ 21 марта 2020

У меня есть парсер, который рекурсивно пересекает дерево. Я хотел бы переписать реализацию этого анализатора таким образом, чтобы я мог использовать Stream API Java и только Stream API без какой-либо рекурсии.

Каждый узел дерева должен обрабатываться процессором со следующей подписью: Stream<Token> process(Token). ИМХО Я не могу использовать flatMap(), потому что я не знаю глубины дерева, которое я обрабатываю, и не могу изменить поток, который я обрабатываю. Я знаю, как сделать это с обычным l oop, простым списком и большим количеством вычислений индекса и смещения.

1 Ответ

0 голосов
/ 21 марта 2020

Если я ясно понимаю:

  1. вам нужен поток всех узлов в вашем дереве. Это правильно?
  2. У вас есть функция process, возвращающая прямых потомков узла

Если это так, и вы действительно хотите исключить любую рекурсию, Spliterator API - это путь к go. Вы создаете потоковый итератор для поглощения элементов в потоке. Любой стандартный Java Stream использует Spliterator под капотом.

EDIT : См. Реализацию, приведенную в конце примера кода (просмотр в глубину).

Однако Если вы разрешите рекурсию, то будет легко создать поток в глубину. В простом примере ниже показаны оба подхода: 

import java.util.Arrays;
import java.util.Spliterator;
import java.util.Stack;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

public class Main {

    /**
     * Sample in-memory node class
     */
    static class Node {
        Node[] children;
        final String name;

        Node(String name) {
            this.name = name;
        }

        @Override
        public String toString() {
            return name;
        }
    }

    /**
     * Simple implementation of your process function, giving back direct children.
     * @param node Node to get children for.
     * @return A stream of available children, can be empty.
     */
    static Stream<Node> process(Node node) {
        return node.children == null? Stream.empty() : Arrays.stream(node.children);
    }

    /**
     * Recursion entrypoint. Allow to perform depth-first browsing.
     */
    static Stream<Node> recurse(Node input) {
        if (input.children == null || input.children.length < 1) return Stream.of(input);
        else return Stream.concat(
                Stream.of(input),
                Arrays.stream(input.children)
                        .flatMap(Main::recurse)
        );
    }

    public static void main(String[] args) throws InterruptedException {

        /* Simple tree :
         * ROOT
         * - A
         * -- AA
         * -- AB
         * --- ABA
         * -- AC
         * - B
         * -- BA
         * -- BB
         * - C
         */
        Node root = new Node("ROOT");
        Node a = new Node("A");
        Node b = new Node("B");
        Node c = new Node("C");
        root.children = new Node[]{a, b, c};

        Node aa = new Node("AA");
        Node ab = new Node("AB");
        Node ac = new Node("AC");
        a.children = new Node[]{aa, ab, ac};

        Node ba = new Node("BA");
        Node bb = new Node("BB");
        b.children = new Node[]{ba, bb};

        Node aba = new Node("ABA");
        ab.children = new Node[]{aba};

        System.out.println("RECURSIVE BROWSING");
        Stream.of(root).flatMap(Main::recurse)
                .forEach(System.out::println);

        System.out.println("NON-RECURSIVE USING SINGLE-THREAD SPLITERATOR");
        final NodeSpliterator spliterator = new NodeSpliterator(root);
        StreamSupport.stream(spliterator, false)
                .forEach(System.out::println);
    }

    static class NodeSpliterator implements Spliterator<Node> {

        private final Stack<Node> nodeQueue = new Stack<>();

        NodeSpliterator(Node root) {
            nodeQueue.push(root);
        }

        @Override
        public boolean tryAdvance(Consumer<? super Node> action) {
            if (nodeQueue.empty()) return false; // No more nodes, end of this fork of the stream.
            final Node next = nodeQueue.pop();
            // Iterate on next available node
            action.accept(next);
            // Sink direct children for next iteration
            final int endOfQueue = nodeQueue.size();
            process(next).forEach(node -> nodeQueue.add(endOfQueue, node));
            // Notify Stream API that an element has successfully been processed.
            return true;
        }

        @Override
        public Spliterator<Node> trySplit() {
            // Little trick : Parallelism of ordered streams requires the fork to distribute a prefix of
            // Available elements. I'll keep things simple here, I let you work on that.
            return null;
        }

        @Override
        public long estimateSize() {
            return Long.MAX_VALUE; // Would require external metadata for estimation, or womplete browsing for exact count
        }

        @Override
        public int characteristics() {
            // Note : you can add Spliterator.CONCURRENT below once you've resolved how to implement trySplit.
            return ORDERED|NONNULL;
        }
    }
}

Этот код выдает следующий вывод: 

RECURSIVE BROWSING
ROOT
A
AA
AB
ABA
AC
B
BA
BB
C
NON-RECURSIVE USING SINGLE-THREAD SPLITERATOR
ROOT
A
AA
AB
ABA
AC
B
BA
BB
C
...