<input id="ohw05"></input>
  • <table id="ohw05"><menu id="ohw05"></menu></table>
  • <var id="ohw05"></var>
  • <code id="ohw05"><cite id="ohw05"></cite></code>
    <label id="ohw05"></label>
    <var id="ohw05"></var>
  • 談談stream的運行原理

      害,別誤會,我這里說的stream不是流式編程,不是大數據處理框架。我這里說的是stream指的是jdk中的一個開發工具包stream. 該工具包在jdk8中出現,可以說已經是冷飯了,為何還要你說?只因各家一言,不算得自家理解,如若有空,何多聽一版又何妨。

      本篇主要從幾個方面講講:1. 我們常見的stream都有哪些? 2. stream包有哪些好處? 3. stream包的實現原理? 相信這些多少會解開大家的一些迷惑。

     

    1. 我們常見的stream都有哪些? 

      stream直接翻譯為流。何謂流?我們最常見的,比如網絡中的數據傳輸,即tcp/udp那一套東西,都是建立在二進制流的基礎上的。用流來形容這些數據或文件的傳輸,非常形象,因為數據總是源源不斷地從一端流向另一端,這是不流是什么。只是,傳輸到另一端之后,我們再做解析,便有了數據或文件之說。其實這說的,便是高層協議了。

      另說一個stream, 那就是jdk中的各種InputStream了,它用于讀取文件數據,讀取byte數據,其實也是源源不斷將數據從一個設備流入到另一設備。jdk中有InputStream/OutputStream, 作為根基,其上則是各種 FileInputStream, FileOutputStream, FileReader, FileWriter,... 實際上,整個io包幾乎都是在圍繞流這個概念來展開的。可見,io是相當的重要啊。

      再說一stream, 則是對大數據的處理了,stream,即是實時數據處理的重要技術實現,因與實時二字吻合,恰好又類似于數據從一設備流入另一設備,且是實時的。所以,stream在大數據領域也是大放異彩啊!比如 spark, flink 你可知?比如 圖數據庫語言標準 gremlin 的算子。

      還有更多的流概念,更多的流實現,不必細說,也無法細說。單只知道,流無處不在,非常重要。

      還有本文要議的stream包,到底是何生物,且看后續說來。

     

    2. stream包有何好處?

      stream包,在java中是以一個工具包的形式存在,即你用則以,不用亦可。

      那么,用它到底有何好處?好處主要有二: 1.可以減少冗余代碼的編寫;比如要寫一個過濾器則只需調用一filter()傳入處理邏輯即可; 2.可以很方便的利用一些隱藏的升級好處或者多核帶來的好處;(當然你可能用不上這些好處)

      說實話,這兩個功能,看起來實際沒有太多的誘惑力,但凡我們封裝幾個方法,供外部調用,不也可以達到同等效果?是了!如果你有這等造詣,能夠抽象出足夠通用的方法,供各方使用,那你不算大牛何人算?說到底,stream也就是高手封裝的工具包而已。

      來幾個應用實例,看看stream都如何使用的:

    public class StreamUtilTest {
    
        @Test
        public void testArrayStream() {
            // 1. 過濾值;改變值;排序;
            Integer[] intArr = {1, 2, 3, 5, 22, 8, 5};
            List<Integer> iArrList = Arrays.stream(intArr)
                                    .filter(r -> r < 20)
                                    .map(r -> r + 1)
                                    .sorted().collect(Collectors.toList());
            System.out.println("result:" + iArrList);
    
            String[] strArr = {"a1,a2", "q,y,h", "ddd,bb,n", null};
            // 2. 過濾數組;拆分值;輸出;
            Arrays.stream(strArr).filter(Objects::nonNull)
                    .flatMap(r -> Arrays.stream(r.split(",")))
                    .forEach(System.out::println);
        }
    
        @Test
        public void testListStream() {
            List<String> list = new ArrayList<>();
            list.add("ab");
            list.add("ccc");
            list.add("ddd");
            // 3. 求list中的最大值
            Optional<String> maxStr = list.stream().max(Comparator.naturalOrder());
            System.out.println(maxStr);
        }
    }

      害,不必糾結里面干的事情復不復雜,有沒有意義,只知道有這用法即可。 反正就當你會這么用,即能解決這般問題。這也是我們高級語言使用必備技能,學會調用api.

      不過需要說明的,java中有一句老話,叫做萬事萬物皆對象。 但見上面的寫法,自然不太像對象。是了,這是lamda語法,雖說另一主題,但何妨在此處一題。但既然說到這,不妨來想想這lamda到底是何物?從某種角度來說,它可以看作是一種內部類,不過寫法不太一樣。但是當我們仔細觀察class文件的變化情況時,發現它與內部類又不太一致,因為java的內部類會在class中生成$xx.class的類文件,而lamda表達式卻不會。但是不管怎么樣,它是可以使用內部類的表達方式獲得同樣的效果,只需將該類代入到其中,即可達到同樣的效果。

      但要細說lamda表達式,則可以反編譯下class文件,可以見些許端倪。

    # 調用lamda表達式示例
    ...
            59: invokestatic  #4                  // Method java/util/Arrays.stream:([Ljava/lang/Object;)Ljava/util/stream/Stream;
            62: invokedynamic #5,  0              // InvokeDynamic #0:test:()Ljava/util/function/Predicate;
            67: invokeinterface #6,  2            // InterfaceMethod java/util/stream/Stream.filter:(Ljava/util/function/Predicate;)Ljava/ut
    il/stream/Stream;
            72: invokedynamic #7,  0              // InvokeDynamic #1:apply:()Ljava/util/function/Function;
            77: invokeinterface #8,  2            // InterfaceMethod java/util/stream/Stream.map:(Ljava/util/function/Function;)Ljava/util/s
    tream/Stream;
    ...
    # 常量池定義,實際是定義了lamda的實現方式為 #0 號方法
        #5 = InvokeDynamic      #0:#102       // #0:test:()Ljava/util/function/Predicate;
    # lamda表達式的具體實現1示例
    BootstrapMethods:
      0: #98 invokestatic java/lang/invoke/LambdaMetafactory.metafactory:(Ljava/lang/invoke/MethodHandles$Lookup;Ljava/lang/String;Ljava/lan
    g/invoke/MethodType;Ljava/lang/invoke/MethodType;Ljava/lang/invoke/MethodHandle;Ljava/lang/invoke/MethodType;)Ljava/lang/invoke/CallSite
    ;
        Method arguments:
          #99 (Ljava/lang/Object;)Z
          // 此處為調用具體的實現方法
          #100 invokestatic com/my/test/common/util/StreamUtilTest.lambda$testArrayStream$0:(Ljava/lang/Integer;)Z
          #101 (Ljava/lang/Integer;)Z
      1: #98 invokestatic java/lang/invoke/LambdaMetafactory.metafactory:(Ljava/lang/invoke/MethodHandles$Lookup;Ljava/lang/String;Ljava/lang/invoke/MethodType;Ljava/lang/invoke/MethodType;Ljava/lang/invoke/MethodHandle;Ljava/lang/invoke/MethodType;)Ljava/lang/invoke/CallSite;
        Method arguments:
          #105 (Ljava/lang/Object;)Ljava/lang/Object;
          #106 invokestatic com/my/test/common/util/StreamUtilTest.lambda$testArrayStream$1:(Ljava/lang/Integer;)Ljava/lang/Integer;
          #107 (Ljava/lang/Integer;)Ljava/lang/Integer;
    
    # lamda表達式具體實現2, 上一步的靜態調用
      private static boolean lambda$testArrayStream$0(java.lang.Integer);
        descriptor: (Ljava/lang/Integer;)Z
        flags: ACC_PRIVATE, ACC_STATIC, ACC_SYNTHETIC
        Code:
          stack=2, locals=1, args_size=1
             0: aload_0
             1: invokevirtual #48                 // Method java/lang/Integer.intValue:()I
             4: bipush        20
             6: if_icmpge     13
             9: iconst_1
            10: goto          14
            13: iconst_0
            14: ireturn
          LineNumberTable:
            line 16: 0
          LocalVariableTable:
            Start  Length  Slot  Name   Signature
                0      15     0     r   Ljava/lang/Integer;
          StackMapTable: number_of_entries = 2
            frame_type = 13 /* same */
            frame_type = 64 /* same_locals_1_stack_item */
              stack = [ int ]
        MethodParameters:
          Name                           Flags
          r                              synthetic

      害,往深了就不說了。單說這lamda表達式,并非使用內部類來實現的,而是使用內部靜態函數來做的,所以也叫函數式編程呢。煩話休提。

      最后,再來看看,這stream包究竟有何神圣地方?其實,就是一個以一個 Stream 接口定義為核心展開的,且看如下:

    /**
     * A sequence of elements supporting sequential and parallel aggregate
     * operations.  The following example illustrates an aggregate operation using
     * {@link Stream} and {@link IntStream}:
     *
     * <pre>{@code
     *     int sum = widgets.stream()
     *                      .filter(w -> w.getColor() == RED)
     *                      .mapToInt(w -> w.getWeight())
     *                      .sum();
     * }</pre>
     *
     * In this example, {@code widgets} is a {@code Collection<Widget>}.  We create
     * a stream of {@code Widget} objects via {@link Collection#stream Collection.stream()},
     * filter it to produce a stream containing only the red widgets, and then
     * transform it into a stream of {@code int} values representing the weight of
     * each red widget. Then this stream is summed to produce a total weight.
     *
     * <p>In addition to {@code Stream}, which is a stream of object references,
     * there are primitive specializations for {@link IntStream}, {@link LongStream},
     * and {@link DoubleStream}, all of which are referred to as "streams" and
     * conform to the characteristics and restrictions described here.
     *
     * <p>To perform a computation, stream
     * <a href="package-summary.html#StreamOps">operations</a> are composed into a
     * <em>stream pipeline</em>.  A stream pipeline consists of a source (which
     * might be an array, a collection, a generator function, an I/O channel,
     * etc), zero or more <em>intermediate operations</em> (which transform a
     * stream into another stream, such as {@link Stream#filter(Predicate)}), and a
     * <em>terminal operation</em> (which produces a result or side-effect, such
     * as {@link Stream#count()} or {@link Stream#forEach(Consumer)}).
     * Streams are lazy; computation on the source data is only performed when the
     * terminal operation is initiated, and source elements are consumed only
     * as needed.
     *
     * <p>Collections and streams, while bearing some superficial similarities,
     * have different goals.  Collections are primarily concerned with the efficient
     * management of, and access to, their elements.  By contrast, streams do not
     * provide a means to directly access or manipulate their elements, and are
     * instead concerned with declaratively describing their source and the
     * computational operations which will be performed in aggregate on that source.
     * However, if the provided stream operations do not offer the desired
     * functionality, the {@link #iterator()} and {@link #spliterator()} operations
     * can be used to perform a controlled traversal.
     *
     * <p>A stream pipeline, like the "widgets" example above, can be viewed as
     * a <em>query</em> on the stream source.  Unless the source was explicitly
     * designed for concurrent modification (such as a {@link ConcurrentHashMap}),
     * unpredictable or erroneous behavior may result from modifying the stream
     * source while it is being queried.
     *
     * <p>Most stream operations accept parameters that describe user-specified
     * behavior, such as the lambda expression {@code w -> w.getWeight()} passed to
     * {@code mapToInt} in the example above.  To preserve correct behavior,
     * these <em>behavioral parameters</em>:
     * <ul>
     * <li>must be <a href="package-summary.html#NonInterference">non-interfering</a>
     * (they do not modify the stream source); and</li>
     * <li>in most cases must be <a href="package-summary.html#Statelessness">stateless</a>
     * (their result should not depend on any state that might change during execution
     * of the stream pipeline).</li>
     * </ul>
     *
     * <p>Such parameters are always instances of a
     * <a href="../function/package-summary.html">functional interface</a> such
     * as {@link java.util.function.Function}, and are often lambda expressions or
     * method references.  Unless otherwise specified these parameters must be
     * <em>non-null</em>.
     *
     * <p>A stream should be operated on (invoking an intermediate or terminal stream
     * operation) only once.  This rules out, for example, "forked" streams, where
     * the same source feeds two or more pipelines, or multiple traversals of the
     * same stream.  A stream implementation may throw {@link IllegalStateException}
     * if it detects that the stream is being reused. However, since some stream
     * operations may return their receiver rather than a new stream object, it may
     * not be possible to detect reuse in all cases.
     *
     * <p>Streams have a {@link #close()} method and implement {@link AutoCloseable},
     * but nearly all stream instances do not actually need to be closed after use.
     * Generally, only streams whose source is an IO channel (such as those returned
     * by {@link Files#lines(Path, Charset)}) will require closing.  Most streams
     * are backed by collections, arrays, or generating functions, which require no
     * special resource management.  (If a stream does require closing, it can be
     * declared as a resource in a {@code try}-with-resources statement.)
     *
     * <p>Stream pipelines may execute either sequentially or in
     * <a href="package-summary.html#Parallelism">parallel</a>.  This
     * execution mode is a property of the stream.  Streams are created
     * with an initial choice of sequential or parallel execution.  (For example,
     * {@link Collection#stream() Collection.stream()} creates a sequential stream,
     * and {@link Collection#parallelStream() Collection.parallelStream()} creates
     * a parallel one.)  This choice of execution mode may be modified by the
     * {@link #sequential()} or {@link #parallel()} methods, and may be queried with
     * the {@link #isParallel()} method.
     *
     * @param <T> the type of the stream elements
     * @since 1.8
     * @see IntStream
     * @see LongStream
     * @see DoubleStream
     * @see <a href="package-summary.html">java.util.stream</a>
     */
    public interface Stream<T> extends BaseStream<T, Stream<T>> {
    
        /**
         * Returns a stream consisting of the elements of this stream that match
         * the given predicate.
         *
         * <p>This is an <a href="package-summary.html#StreamOps">intermediate
         * operation</a>.
         *
         * @param predicate a <a href="package-summary.html#NonInterference">non-interfering</a>,
         *                  <a href="package-summary.html#Statelessness">stateless</a>
         *                  predicate to apply to each element to determine if it
         *                  should be included
         * @return the new stream
         */
        Stream<T> filter(Predicate<? super T> predicate);
    
        /**
         * Returns a stream consisting of the results of applying the given
         * function to the elements of this stream.
         *
         * <p>This is an <a href="package-summary.html#StreamOps">intermediate
         * operation</a>.
         *
         * @param <R> The element type of the new stream
         * @param mapper a <a href="package-summary.html#NonInterference">non-interfering</a>,
         *               <a href="package-summary.html#Statelessness">stateless</a>
         *               function to apply to each element
         * @return the new stream
         */
        <R> Stream<R> map(Function<? super T, ? extends R> mapper);
    
        /**
         * Returns an {@code IntStream} consisting of the results of applying the
         * given function to the elements of this stream.
         *
         * <p>This is an <a href="package-summary.html#StreamOps">
         *     intermediate operation</a>.
         *
         * @param mapper a <a href="package-summary.html#NonInterference">non-interfering</a>,
         *               <a href="package-summary.html#Statelessness">stateless</a>
         *               function to apply to each element
         * @return the new stream
         */
        IntStream mapToInt(ToIntFunction<? super T> mapper);
    
        /**
         * Returns a {@code LongStream} consisting of the results of applying the
         * given function to the elements of this stream.
         *
         * <p>This is an <a href="package-summary.html#StreamOps">intermediate
         * operation</a>.
         *
         * @param mapper a <a href="package-summary.html#NonInterference">non-interfering</a>,
         *               <a href="package-summary.html#Statelessness">stateless</a>
         *               function to apply to each element
         * @return the new stream
         */
        LongStream mapToLong(ToLongFunction<? super T> mapper);
    
        /**
         * Returns a {@code DoubleStream} consisting of the results of applying the
         * given function to the elements of this stream.
         *
         * <p>This is an <a href="package-summary.html#StreamOps">intermediate
         * operation</a>.
         *
         * @param mapper a <a href="package-summary.html#NonInterference">non-interfering</a>,
         *               <a href="package-summary.html#Statelessness">stateless</a>
         *               function to apply to each element
         * @return the new stream
         */
        DoubleStream mapToDouble(ToDoubleFunction<? super T> mapper);
    
        /**
         * Returns a stream consisting of the results of replacing each element of
         * this stream with the contents of a mapped stream produced by applying
         * the provided mapping function to each element.  Each mapped stream is
         * {@link java.util.stream.BaseStream#close() closed} after its contents
         * have been placed into this stream.  (If a mapped stream is {@code null}
         * an empty stream is used, instead.)
         *
         * <p>This is an <a href="package-summary.html#StreamOps">intermediate
         * operation</a>.
         *
         * @apiNote
         * The {@code flatMap()} operation has the effect of applying a one-to-many
         * transformation to the elements of the stream, and then flattening the
         * resulting elements into a new stream.
         *
         * <p><b>Examples.</b>
         *
         * <p>If {@code orders} is a stream of purchase orders, and each purchase
         * order contains a collection of line items, then the following produces a
         * stream containing all the line items in all the orders:
         * <pre>{@code
         *     orders.flatMap(order -> order.getLineItems().stream())...
         * }</pre>
         *
         * <p>If {@code path} is the path to a file, then the following produces a
         * stream of the {@code words} contained in that file:
         * <pre>{@code
         *     Stream<String> lines = Files.lines(path, StandardCharsets.UTF_8);
         *     Stream<String> words = lines.flatMap(line -> Stream.of(line.split(" +")));
         * }</pre>
         * The {@code mapper} function passed to {@code flatMap} splits a line,
         * using a simple regular expression, into an array of words, and then
         * creates a stream of words from that array.
         *
         * @param <R> The element type of the new stream
         * @param mapper a <a href="package-summary.html#NonInterference">non-interfering</a>,
         *               <a href="package-summary.html#Statelessness">stateless</a>
         *               function to apply to each element which produces a stream
         *               of new values
         * @return the new stream
         */
        <R> Stream<R> flatMap(Function<? super T, ? extends Stream<? extends R>> mapper);
    
        /**
         * Returns an {@code IntStream} consisting of the results of replacing each
         * element of this stream with the contents of a mapped stream produced by
         * applying the provided mapping function to each element.  Each mapped
         * stream is {@link java.util.stream.BaseStream#close() closed} after its
         * contents have been placed into this stream.  (If a mapped stream is
         * {@code null} an empty stream is used, instead.)
         *
         * <p>This is an <a href="package-summary.html#StreamOps">intermediate
         * operation</a>.
         *
         * @param mapper a <a href="package-summary.html#NonInterference">non-interfering</a>,
         *               <a href="package-summary.html#Statelessness">stateless</a>
         *               function to apply to each element which produces a stream
         *               of new values
         * @return the new stream
         * @see #flatMap(Function)
         */
        IntStream flatMapToInt(Function<? super T, ? extends IntStream> mapper);
    
        /**
         * Returns an {@code LongStream} consisting of the results of replacing each
         * element of this stream with the contents of a mapped stream produced by
         * applying the provided mapping function to each element.  Each mapped
         * stream is {@link java.util.stream.BaseStream#close() closed} after its
         * contents have been placed into this stream.  (If a mapped stream is
         * {@code null} an empty stream is used, instead.)
         *
         * <p>This is an <a href="package-summary.html#StreamOps">intermediate
         * operation</a>.
         *
         * @param mapper a <a href="package-summary.html#NonInterference">non-interfering</a>,
         *               <a href="package-summary.html#Statelessness">stateless</a>
         *               function to apply to each element which produces a stream
         *               of new values
         * @return the new stream
         * @see #flatMap(Function)
         */
        LongStream flatMapToLong(Function<? super T, ? extends LongStream> mapper);
    
        /**
         * Returns an {@code DoubleStream} consisting of the results of replacing
         * each element of this stream with the contents of a mapped stream produced
         * by applying the provided mapping function to each element.  Each mapped
         * stream is {@link java.util.stream.BaseStream#close() closed} after its
         * contents have placed been into this stream.  (If a mapped stream is
         * {@code null} an empty stream is used, instead.)
         *
         * <p>This is an <a href="package-summary.html#StreamOps">intermediate
         * operation</a>.
         *
         * @param mapper a <a href="package-summary.html#NonInterference">non-interfering</a>,
         *               <a href="package-summary.html#Statelessness">stateless</a>
         *               function to apply to each element which produces a stream
         *               of new values
         * @return the new stream
         * @see #flatMap(Function)
         */
        DoubleStream flatMapToDouble(Function<? super T, ? extends DoubleStream> mapper);
    
        /**
         * Returns a stream consisting of the distinct elements (according to
         * {@link Object#equals(Object)}) of this stream.
         *
         * <p>For ordered streams, the selection of distinct elements is stable
         * (for duplicated elements, the element appearing first in the encounter
         * order is preserved.)  For unordered streams, no stability guarantees
         * are made.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">stateful
         * intermediate operation</a>.
         *
         * @apiNote
         * Preserving stability for {@code distinct()} in parallel pipelines is
         * relatively expensive (requires that the operation act as a full barrier,
         * with substantial buffering overhead), and stability is often not needed.
         * Using an unordered stream source (such as {@link #generate(Supplier)})
         * or removing the ordering constraint with {@link #unordered()} may result
         * in significantly more efficient execution for {@code distinct()} in parallel
         * pipelines, if the semantics of your situation permit.  If consistency
         * with encounter order is required, and you are experiencing poor performance
         * or memory utilization with {@code distinct()} in parallel pipelines,
         * switching to sequential execution with {@link #sequential()} may improve
         * performance.
         *
         * @return the new stream
         */
        Stream<T> distinct();
    
        /**
         * Returns a stream consisting of the elements of this stream, sorted
         * according to natural order.  If the elements of this stream are not
         * {@code Comparable}, a {@code java.lang.ClassCastException} may be thrown
         * when the terminal operation is executed.
         *
         * <p>For ordered streams, the sort is stable.  For unordered streams, no
         * stability guarantees are made.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">stateful
         * intermediate operation</a>.
         *
         * @return the new stream
         */
        Stream<T> sorted();
    
        /**
         * Returns a stream consisting of the elements of this stream, sorted
         * according to the provided {@code Comparator}.
         *
         * <p>For ordered streams, the sort is stable.  For unordered streams, no
         * stability guarantees are made.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">stateful
         * intermediate operation</a>.
         *
         * @param comparator a <a href="package-summary.html#NonInterference">non-interfering</a>,
         *                   <a href="package-summary.html#Statelessness">stateless</a>
         *                   {@code Comparator} to be used to compare stream elements
         * @return the new stream
         */
        Stream<T> sorted(Comparator<? super T> comparator);
    
        /**
         * Returns a stream consisting of the elements of this stream, additionally
         * performing the provided action on each element as elements are consumed
         * from the resulting stream.
         *
         * <p>This is an <a href="package-summary.html#StreamOps">intermediate
         * operation</a>.
         *
         * <p>For parallel stream pipelines, the action may be called at
         * whatever time and in whatever thread the element is made available by the
         * upstream operation.  If the action modifies shared state,
         * it is responsible for providing the required synchronization.
         *
         * @apiNote This method exists mainly to support debugging, where you want
         * to see the elements as they flow past a certain point in a pipeline:
         * <pre>{@code
         *     Stream.of("one", "two", "three", "four")
         *         .filter(e -> e.length() > 3)
         *         .peek(e -> System.out.println("Filtered value: " + e))
         *         .map(String::toUpperCase)
         *         .peek(e -> System.out.println("Mapped value: " + e))
         *         .collect(Collectors.toList());
         * }</pre>
         *
         * @param action a <a href="package-summary.html#NonInterference">
         *                 non-interfering</a> action to perform on the elements as
         *                 they are consumed from the stream
         * @return the new stream
         */
        Stream<T> peek(Consumer<? super T> action);
    
        /**
         * Returns a stream consisting of the elements of this stream, truncated
         * to be no longer than {@code maxSize} in length.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">short-circuiting
         * stateful intermediate operation</a>.
         *
         * @apiNote
         * While {@code limit()} is generally a cheap operation on sequential
         * stream pipelines, it can be quite expensive on ordered parallel pipelines,
         * especially for large values of {@code maxSize}, since {@code limit(n)}
         * is constrained to return not just any <em>n</em> elements, but the
         * <em>first n</em> elements in the encounter order.  Using an unordered
         * stream source (such as {@link #generate(Supplier)}) or removing the
         * ordering constraint with {@link #unordered()} may result in significant
         * speedups of {@code limit()} in parallel pipelines, if the semantics of
         * your situation permit.  If consistency with encounter order is required,
         * and you are experiencing poor performance or memory utilization with
         * {@code limit()} in parallel pipelines, switching to sequential execution
         * with {@link #sequential()} may improve performance.
         *
         * @param maxSize the number of elements the stream should be limited to
         * @return the new stream
         * @throws IllegalArgumentException if {@code maxSize} is negative
         */
        Stream<T> limit(long maxSize);
    
        /**
         * Returns a stream consisting of the remaining elements of this stream
         * after discarding the first {@code n} elements of the stream.
         * If this stream contains fewer than {@code n} elements then an
         * empty stream will be returned.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">stateful
         * intermediate operation</a>.
         *
         * @apiNote
         * While {@code skip()} is generally a cheap operation on sequential
         * stream pipelines, it can be quite expensive on ordered parallel pipelines,
         * especially for large values of {@code n}, since {@code skip(n)}
         * is constrained to skip not just any <em>n</em> elements, but the
         * <em>first n</em> elements in the encounter order.  Using an unordered
         * stream source (such as {@link #generate(Supplier)}) or removing the
         * ordering constraint with {@link #unordered()} may result in significant
         * speedups of {@code skip()} in parallel pipelines, if the semantics of
         * your situation permit.  If consistency with encounter order is required,
         * and you are experiencing poor performance or memory utilization with
         * {@code skip()} in parallel pipelines, switching to sequential execution
         * with {@link #sequential()} may improve performance.
         *
         * @param n the number of leading elements to skip
         * @return the new stream
         * @throws IllegalArgumentException if {@code n} is negative
         */
        Stream<T> skip(long n);
    
        /**
         * Performs an action for each element of this stream.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">terminal
         * operation</a>.
         *
         * <p>The behavior of this operation is explicitly nondeterministic.
         * For parallel stream pipelines, this operation does <em>not</em>
         * guarantee to respect the encounter order of the stream, as doing so
         * would sacrifice the benefit of parallelism.  For any given element, the
         * action may be performed at whatever time and in whatever thread the
         * library chooses.  If the action accesses shared state, it is
         * responsible for providing the required synchronization.
         *
         * @param action a <a href="package-summary.html#NonInterference">
         *               non-interfering</a> action to perform on the elements
         */
        void forEach(Consumer<? super T> action);
    
        /**
         * Performs an action for each element of this stream, in the encounter
         * order of the stream if the stream has a defined encounter order.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">terminal
         * operation</a>.
         *
         * <p>This operation processes the elements one at a time, in encounter
         * order if one exists.  Performing the action for one element
         * <a href="../concurrent/package-summary.html#MemoryVisibility"><i>happens-before</i></a>
         * performing the action for subsequent elements, but for any given element,
         * the action may be performed in whatever thread the library chooses.
         *
         * @param action a <a href="package-summary.html#NonInterference">
         *               non-interfering</a> action to perform on the elements
         * @see #forEach(Consumer)
         */
        void forEachOrdered(Consumer<? super T> action);
    
        /**
         * Returns an array containing the elements of this stream.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">terminal
         * operation</a>.
         *
         * @return an array containing the elements of this stream
         */
        Object[] toArray();
    
        /**
         * Returns an array containing the elements of this stream, using the
         * provided {@code generator} function to allocate the returned array, as
         * well as any additional arrays that might be required for a partitioned
         * execution or for resizing.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">terminal
         * operation</a>.
         *
         * @apiNote
         * The generator function takes an integer, which is the size of the
         * desired array, and produces an array of the desired size.  This can be
         * concisely expressed with an array constructor reference:
         * <pre>{@code
         *     Person[] men = people.stream()
         *                          .filter(p -> p.getGender() == MALE)
         *                          .toArray(Person[]::new);
         * }</pre>
         *
         * @param <A> the element type of the resulting array
         * @param generator a function which produces a new array of the desired
         *                  type and the provided length
         * @return an array containing the elements in this stream
         * @throws ArrayStoreException if the runtime type of the array returned
         * from the array generator is not a supertype of the runtime type of every
         * element in this stream
         */
        <A> A[] toArray(IntFunction<A[]> generator);
    
        /**
         * Performs a <a href="package-summary.html#Reduction">reduction</a> on the
         * elements of this stream, using the provided identity value and an
         * <a href="package-summary.html#Associativity">associative</a>
         * accumulation function, and returns the reduced value.  This is equivalent
         * to:
         * <pre>{@code
         *     T result = identity;
         *     for (T element : this stream)
         *         result = accumulator.apply(result, element)
         *     return result;
         * }</pre>
         *
         * but is not constrained to execute sequentially.
         *
         * <p>The {@code identity} value must be an identity for the accumulator
         * function. This means that for all {@code t},
         * {@code accumulator.apply(identity, t)} is equal to {@code t}.
         * The {@code accumulator} function must be an
         * <a href="package-summary.html#Associativity">associative</a> function.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">terminal
         * operation</a>.
         *
         * @apiNote Sum, min, max, average, and string concatenation are all special
         * cases of reduction. Summing a stream of numbers can be expressed as:
         *
         * <pre>{@code
         *     Integer sum = integers.reduce(0, (a, b) -> a+b);
         * }</pre>
         *
         * or:
         *
         * <pre>{@code
         *     Integer sum = integers.reduce(0, Integer::sum);
         * }</pre>
         *
         * <p>While this may seem a more roundabout way to perform an aggregation
         * compared to simply mutating a running total in a loop, reduction
         * operations parallelize more gracefully, without needing additional
         * synchronization and with greatly reduced risk of data races.
         *
         * @param identity the identity value for the accumulating function
         * @param accumulator an <a href="package-summary.html#Associativity">associative</a>,
         *                    <a href="package-summary.html#NonInterference">non-interfering</a>,
         *                    <a href="package-summary.html#Statelessness">stateless</a>
         *                    function for combining two values
         * @return the result of the reduction
         */
        T reduce(T identity, BinaryOperator<T> accumulator);
    
        /**
         * Performs a <a href="package-summary.html#Reduction">reduction</a> on the
         * elements of this stream, using an
         * <a href="package-summary.html#Associativity">associative</a> accumulation
         * function, and returns an {@code Optional} describing the reduced value,
         * if any. This is equivalent to:
         * <pre>{@code
         *     boolean foundAny = false;
         *     T result = null;
         *     for (T element : this stream) {
         *         if (!foundAny) {
         *             foundAny = true;
         *             result = element;
         *         }
         *         else
         *             result = accumulator.apply(result, element);
         *     }
         *     return foundAny ? Optional.of(result) : Optional.empty();
         * }</pre>
         *
         * but is not constrained to execute sequentially.
         *
         * <p>The {@code accumulator} function must be an
         * <a href="package-summary.html#Associativity">associative</a> function.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">terminal
         * operation</a>.
         *
         * @param accumulator an <a href="package-summary.html#Associativity">associative</a>,
         *                    <a href="package-summary.html#NonInterference">non-interfering</a>,
         *                    <a href="package-summary.html#Statelessness">stateless</a>
         *                    function for combining two values
         * @return an {@link Optional} describing the result of the reduction
         * @throws NullPointerException if the result of the reduction is null
         * @see #reduce(Object, BinaryOperator)
         * @see #min(Comparator)
         * @see #max(Comparator)
         */
        Optional<T> reduce(BinaryOperator<T> accumulator);
    
        /**
         * Performs a <a href="package-summary.html#Reduction">reduction</a> on the
         * elements of this stream, using the provided identity, accumulation and
         * combining functions.  This is equivalent to:
         * <pre>{@code
         *     U result = identity;
         *     for (T element : this stream)
         *         result = accumulator.apply(result, element)
         *     return result;
         * }</pre>
         *
         * but is not constrained to execute sequentially.
         *
         * <p>The {@code identity} value must be an identity for the combiner
         * function.  This means that for all {@code u}, {@code combiner(identity, u)}
         * is equal to {@code u}.  Additionally, the {@code combiner} function
         * must be compatible with the {@code accumulator} function; for all
         * {@code u} and {@code t}, the following must hold:
         * <pre>{@code
         *     combiner.apply(u, accumulator.apply(identity, t)) == accumulator.apply(u, t)
         * }</pre>
         *
         * <p>This is a <a href="package-summary.html#StreamOps">terminal
         * operation</a>.
         *
         * @apiNote Many reductions using this form can be represented more simply
         * by an explicit combination of {@code map} and {@code reduce} operations.
         * The {@code accumulator} function acts as a fused mapper and accumulator,
         * which can sometimes be more efficient than separate mapping and reduction,
         * such as when knowing the previously reduced value allows you to avoid
         * some computation.
         *
         * @param <U> The type of the result
         * @param identity the identity value for the combiner function
         * @param accumulator an <a href="package-summary.html#Associativity">associative</a>,
         *                    <a href="package-summary.html#NonInterference">non-interfering</a>,
         *                    <a href="package-summary.html#Statelessness">stateless</a>
         *                    function for incorporating an additional element into a result
         * @param combiner an <a href="package-summary.html#Associativity">associative</a>,
         *                    <a href="package-summary.html#NonInterference">non-interfering</a>,
         *                    <a href="package-summary.html#Statelessness">stateless</a>
         *                    function for combining two values, which must be
         *                    compatible with the accumulator function
         * @return the result of the reduction
         * @see #reduce(BinaryOperator)
         * @see #reduce(Object, BinaryOperator)
         */
        <U> U reduce(U identity,
                     BiFunction<U, ? super T, U> accumulator,
                     BinaryOperator<U> combiner);
    
        /**
         * Performs a <a href="package-summary.html#MutableReduction">mutable
         * reduction</a> operation on the elements of this stream.  A mutable
         * reduction is one in which the reduced value is a mutable result container,
         * such as an {@code ArrayList}, and elements are incorporated by updating
         * the state of the result rather than by replacing the result.  This
         * produces a result equivalent to:
         * <pre>{@code
         *     R result = supplier.get();
         *     for (T element : this stream)
         *         accumulator.accept(result, element);
         *     return result;
         * }</pre>
         *
         * <p>Like {@link #reduce(Object, BinaryOperator)}, {@code collect} operations
         * can be parallelized without requiring additional synchronization.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">terminal
         * operation</a>.
         *
         * @apiNote There are many existing classes in the JDK whose signatures are
         * well-suited for use with method references as arguments to {@code collect()}.
         * For example, the following will accumulate strings into an {@code ArrayList}:
         * <pre>{@code
         *     List<String> asList = stringStream.collect(ArrayList::new, ArrayList::add,
         *                                                ArrayList::addAll);
         * }</pre>
         *
         * <p>The following will take a stream of strings and concatenates them into a
         * single string:
         * <pre>{@code
         *     String concat = stringStream.collect(StringBuilder::new, StringBuilder::append,
         *                                          StringBuilder::append)
         *                                 .toString();
         * }</pre>
         *
         * @param <R> type of the result
         * @param supplier a function that creates a new result container. For a
         *                 parallel execution, this function may be called
         *                 multiple times and must return a fresh value each time.
         * @param accumulator an <a href="package-summary.html#Associativity">associative</a>,
         *                    <a href="package-summary.html#NonInterference">non-interfering</a>,
         *                    <a href="package-summary.html#Statelessness">stateless</a>
         *                    function for incorporating an additional element into a result
         * @param combiner an <a href="package-summary.html#Associativity">associative</a>,
         *                    <a href="package-summary.html#NonInterference">non-interfering</a>,
         *                    <a href="package-summary.html#Statelessness">stateless</a>
         *                    function for combining two values, which must be
         *                    compatible with the accumulator function
         * @return the result of the reduction
         */
        <R> R collect(Supplier<R> supplier,
                      BiConsumer<R, ? super T> accumulator,
                      BiConsumer<R, R> combiner);
    
        /**
         * Performs a <a href="package-summary.html#MutableReduction">mutable
         * reduction</a> operation on the elements of this stream using a
         * {@code Collector}.  A {@code Collector}
         * encapsulates the functions used as arguments to
         * {@link #collect(Supplier, BiConsumer, BiConsumer)}, allowing for reuse of
         * collection strategies and composition of collect operations such as
         * multiple-level grouping or partitioning.
         *
         * <p>If the stream is parallel, and the {@code Collector}
         * is {@link Collector.Characteristics#CONCURRENT concurrent}, and
         * either the stream is unordered or the collector is
         * {@link Collector.Characteristics#UNORDERED unordered},
         * then a concurrent reduction will be performed (see {@link Collector} for
         * details on concurrent reduction.)
         *
         * <p>This is a <a href="package-summary.html#StreamOps">terminal
         * operation</a>.
         *
         * <p>When executed in parallel, multiple intermediate results may be
         * instantiated, populated, and merged so as to maintain isolation of
         * mutable data structures.  Therefore, even when executed in parallel
         * with non-thread-safe data structures (such as {@code ArrayList}), no
         * additional synchronization is needed for a parallel reduction.
         *
         * @apiNote
         * The following will accumulate strings into an ArrayList:
         * <pre>{@code
         *     List<String> asList = stringStream.collect(Collectors.toList());
         * }</pre>
         *
         * <p>The following will classify {@code Person} objects by city:
         * <pre>{@code
         *     Map<String, List<Person>> peopleByCity
         *         = personStream.collect(Collectors.groupingBy(Person::getCity));
         * }</pre>
         *
         * <p>The following will classify {@code Person} objects by state and city,
         * cascading two {@code Collector}s together:
         * <pre>{@code
         *     Map<String, Map<String, List<Person>>> peopleByStateAndCity
         *         = personStream.collect(Collectors.groupingBy(Person::getState,
         *                                                      Collectors.groupingBy(Person::getCity)));
         * }</pre>
         *
         * @param <R> the type of the result
         * @param <A> the intermediate accumulation type of the {@code Collector}
         * @param collector the {@code Collector} describing the reduction
         * @return the result of the reduction
         * @see #collect(Supplier, BiConsumer, BiConsumer)
         * @see Collectors
         */
        <R, A> R collect(Collector<? super T, A, R> collector);
    
        /**
         * Returns the minimum element of this stream according to the provided
         * {@code Comparator}.  This is a special case of a
         * <a href="package-summary.html#Reduction">reduction</a>.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">terminal operation</a>.
         *
         * @param comparator a <a href="package-summary.html#NonInterference">non-interfering</a>,
         *                   <a href="package-summary.html#Statelessness">stateless</a>
         *                   {@code Comparator} to compare elements of this stream
         * @return an {@code Optional} describing the minimum element of this stream,
         * or an empty {@code Optional} if the stream is empty
         * @throws NullPointerException if the minimum element is null
         */
        Optional<T> min(Comparator<? super T> comparator);
    
        /**
         * Returns the maximum element of this stream according to the provided
         * {@code Comparator}.  This is a special case of a
         * <a href="package-summary.html#Reduction">reduction</a>.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">terminal
         * operation</a>.
         *
         * @param comparator a <a href="package-summary.html#NonInterference">non-interfering</a>,
         *                   <a href="package-summary.html#Statelessness">stateless</a>
         *                   {@code Comparator} to compare elements of this stream
         * @return an {@code Optional} describing the maximum element of this stream,
         * or an empty {@code Optional} if the stream is empty
         * @throws NullPointerException if the maximum element is null
         */
        Optional<T> max(Comparator<? super T> comparator);
    
        /**
         * Returns the count of elements in this stream.  This is a special case of
         * a <a href="package-summary.html#Reduction">reduction</a> and is
         * equivalent to:
         * <pre>{@code
         *     return mapToLong(e -> 1L).sum();
         * }</pre>
         *
         * <p>This is a <a href="package-summary.html#StreamOps">terminal operation</a>.
         *
         * @return the count of elements in this stream
         */
        long count();
    
        /**
         * Returns whether any elements of this stream match the provided
         * predicate.  May not evaluate the predicate on all elements if not
         * necessary for determining the result.  If the stream is empty then
         * {@code false} is returned and the predicate is not evaluated.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">short-circuiting
         * terminal operation</a>.
         *
         * @apiNote
         * This method evaluates the <em>existential quantification</em> of the
         * predicate over the elements of the stream (for some x P(x)).
         *
         * @param predicate a <a href="package-summary.html#NonInterference">non-interfering</a>,
         *                  <a href="package-summary.html#Statelessness">stateless</a>
         *                  predicate to apply to elements of this stream
         * @return {@code true} if any elements of the stream match the provided
         * predicate, otherwise {@code false}
         */
        boolean anyMatch(Predicate<? super T> predicate);
    
        /**
         * Returns whether all elements of this stream match the provided predicate.
         * May not evaluate the predicate on all elements if not necessary for
         * determining the result.  If the stream is empty then {@code true} is
         * returned and the predicate is not evaluated.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">short-circuiting
         * terminal operation</a>.
         *
         * @apiNote
         * This method evaluates the <em>universal quantification</em> of the
         * predicate over the elements of the stream (for all x P(x)).  If the
         * stream is empty, the quantification is said to be <em>vacuously
         * satisfied</em> and is always {@code true} (regardless of P(x)).
         *
         * @param predicate a <a href="package-summary.html#NonInterference">non-interfering</a>,
         *                  <a href="package-summary.html#Statelessness">stateless</a>
         *                  predicate to apply to elements of this stream
         * @return {@code true} if either all elements of the stream match the
         * provided predicate or the stream is empty, otherwise {@code false}
         */
        boolean allMatch(Predicate<? super T> predicate);
    
        /**
         * Returns whether no elements of this stream match the provided predicate.
         * May not evaluate the predicate on all elements if not necessary for
         * determining the result.  If the stream is empty then {@code true} is
         * returned and the predicate is not evaluated.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">short-circuiting
         * terminal operation</a>.
         *
         * @apiNote
         * This method evaluates the <em>universal quantification</em> of the
         * negated predicate over the elements of the stream (for all x ~P(x)).  If
         * the stream is empty, the quantification is said to be vacuously satisfied
         * and is always {@code true}, regardless of P(x).
         *
         * @param predicate a <a href="package-summary.html#NonInterference">non-interfering</a>,
         *                  <a href="package-summary.html#Statelessness">stateless</a>
         *                  predicate to apply to elements of this stream
         * @return {@code true} if either no elements of the stream match the
         * provided predicate or the stream is empty, otherwise {@code false}
         */
        boolean noneMatch(Predicate<? super T> predicate);
    
        /**
         * Returns an {@link Optional} describing the first element of this stream,
         * or an empty {@code Optional} if the stream is empty.  If the stream has
         * no encounter order, then any element may be returned.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">short-circuiting
         * terminal operation</a>.
         *
         * @return an {@code Optional} describing the first element of this stream,
         * or an empty {@code Optional} if the stream is empty
         * @throws NullPointerException if the element selected is null
         */
        Optional<T> findFirst();
    
        /**
         * Returns an {@link Optional} describing some element of the stream, or an
         * empty {@code Optional} if the stream is empty.
         *
         * <p>This is a <a href="package-summary.html#StreamOps">short-circuiting
         * terminal operation</a>.
         *
         * <p>The behavior of this operation is explicitly nondeterministic; it is
         * free to select any element in the stream.  This is to allow for maximal
         * performance in parallel operations; the cost is that multiple invocations
         * on the same source may not return the same result.  (If a stable result
         * is desired, use {@link #findFirst()} instead.)
         *
         * @return an {@code Optional} describing some element of this stream, or an
         * empty {@code Optional} if the stream is empty
         * @throws NullPointerException if the element selected is null
         * @see #findFirst()
         */
        Optional<T> findAny();
    
        // Static factories
    
        /**
         * Returns a builder for a {@code Stream}.
         *
         * @param <T> type of elements
         * @return a stream builder
         */
        public static<T> Builder<T> builder() {
            return new Streams.StreamBuilderImpl<>();
        }
    
        /**
         * Returns an empty sequential {@code Stream}.
         *
         * @param <T> the type of stream elements
         * @return an empty sequential stream
         */
        public static<T> Stream<T> empty() {
            return StreamSupport.stream(Spliterators.<T>emptySpliterator(), false);
        }
    
        /**
         * Returns a sequential {@code Stream} containing a single element.
         *
         * @param t the single element
         * @param <T> the type of stream elements
         * @return a singleton sequential stream
         */
        public static<T> Stream<T> of(T t) {
            return StreamSupport.stream(new Streams.StreamBuilderImpl<>(t), false);
        }
    
        /**
         * Returns a sequential ordered stream whose elements are the specified values.
         *
         * @param <T> the type of stream elements
         * @param values the elements of the new stream
         * @return the new stream
         */
        @SafeVarargs
        @SuppressWarnings("varargs") // Creating a stream from an array is safe
        public static<T> Stream<T> of(T... values) {
            return Arrays.stream(values);
        }
    
        /**
         * Returns an infinite sequential ordered {@code Stream} produced by iterative
         * application of a function {@code f} to an initial element {@code seed},
         * producing a {@code Stream} consisting of {@code seed}, {@code f(seed)},
         * {@code f(f(seed))}, etc.
         *
         * <p>The first element (position {@code 0}) in the {@code Stream} will be
         * the provided {@code seed}.  For {@code n > 0}, the element at position
         * {@code n}, will be the result of applying the function {@code f} to the
         * element at position {@code n - 1}.
         *
         * @param <T> the type of stream elements
         * @param seed the initial element
         * @param f a function to be applied to to the previous element to produce
         *          a new element
         * @return a new sequential {@code Stream}
         */
        public static<T> Stream<T> iterate(final T seed, final UnaryOperator<T> f) {
            Objects.requireNonNull(f);
            final Iterator<T> iterator = new Iterator<T>() {
                @SuppressWarnings("unchecked")
                T t = (T) Streams.NONE;
    
                @Override
                public boolean hasNext() {
                    return true;
                }
    
                @Override
                public T next() {
                    return t = (t == Streams.NONE) ? seed : f.apply(t);
                }
            };
            return StreamSupport.stream(Spliterators.spliteratorUnknownSize(
                    iterator,
                    Spliterator.ORDERED | Spliterator.IMMUTABLE), false);
        }
    
        /**
         * Returns an infinite sequential unordered stream where each element is
         * generated by the provided {@code Supplier}.  This is suitable for
         * generating constant streams, streams of random elements, etc.
         *
         * @param <T> the type of stream elements
         * @param s the {@code Supplier} of generated elements
         * @return a new infinite sequential unordered {@code Stream}
         */
        public static<T> Stream<T> generate(Supplier<T> s) {
            Objects.requireNonNull(s);
            return StreamSupport.stream(
                    new StreamSpliterators.InfiniteSupplyingSpliterator.OfRef<>(Long.MAX_VALUE, s), false);
        }
    
        /**
         * Creates a lazily concatenated stream whose elements are all the
         * elements of the first stream followed by all the elements of the
         * second stream.  The resulting stream is ordered if both
         * of the input streams are ordered, and parallel if either of the input
         * streams is parallel.  When the resulting stream is closed, the close
         * handlers for both input streams are invoked.
         *
         * @implNote
         * Use caution when constructing streams from repeated concatenation.
         * Accessing an element of a deeply concatenated stream can result in deep
         * call chains, or even {@code StackOverflowException}.
         *
         * @param <T> The type of stream elements
         * @param a the first stream
         * @param b the second stream
         * @return the concatenation of the two input streams
         */
        public static <T> Stream<T> concat(Stream<? extends T> a, Stream<? extends T> b) {
            Objects.requireNonNull(a);
            Objects.requireNonNull(b);
    
            @SuppressWarnings("unchecked")
            Spliterator<T> split = new Streams.ConcatSpliterator.OfRef<>(
                    (Spliterator<T>) a.spliterator(), (Spliterator<T>) b.spliterator());
            Stream<T> stream = StreamSupport.stream(split, a.isParallel() || b.isParallel());
            return stream.onClose(Streams.composedClose(a, b));
        }
    
        /**
         * A mutable builder for a {@code Stream}.  This allows the creation of a
         * {@code Stream} by generating elements individually and adding them to the
         * {@code Builder} (without the copying overhead that comes from using
         * an {@code ArrayList} as a temporary buffer.)
         *
         * <p>A stream builder has a lifecycle, which starts in a building
         * phase, during which elements can be added, and then transitions to a built
         * phase, after which elements may not be added.  The built phase begins
         * when the {@link #build()} method is called, which creates an ordered
         * {@code Stream} whose elements are the elements that were added to the stream
         * builder, in the order they were added.
         *
         * @param <T> the type of stream elements
         * @see Stream#builder()
         * @since 1.8
         */
        public interface Builder<T> extends Consumer<T> {
    
            /**
             * Adds an element to the stream being built.
             *
             * @throws IllegalStateException if the builder has already transitioned to
             * the built state
             */
            @Override
            void accept(T t);
    
            /**
             * Adds an element to the stream being built.
             *
             * @implSpec
             * The default implementation behaves as if:
             * <pre>{@code
             *     accept(t)
             *     return this;
             * }</pre>
             *
             * @param t the element to add
             * @return {@code this} builder
             * @throws IllegalStateException if the builder has already transitioned to
             * the built state
             */
            default Builder<T> add(T t) {
                accept(t);
                return this;
            }
    
            /**
             * Builds the stream, transitioning this builder to the built state.
             * An {@code IllegalStateException} is thrown if there are further attempts
             * to operate on the builder after it has entered the built state.
             *
             * @return the built stream
             * @throws IllegalStateException if the builder has already transitioned to
             * the built state
             */
            Stream<T> build();
    
        }
    }
    View Code

      只是,這接口中定義的參數,都是些經過特殊定義的接口,即函數式接口,即默認只需實現一個方法即可接口類定義。

     

    3. stream包的具體實現?

      如上一節,我們已知stream中主要依賴于許多的接口定義。既然是接口,那就必然無法直接調用,須要有與之對應的實現方可調用。所以,我們需要有特定的場景,才可以來談stream 的實現問題。

      所以,我們先以相對簡單的 Integer 的流轉化與處理過程,一探stream究竟。

         // java.util.Arrays#stream(T[])
        /**
         * Returns a sequential {@link Stream} with the specified array as its
         * source.
         *
         * @param <T> The type of the array elements
         * @param array The array, assumed to be unmodified during use
         * @return a {@code Stream} for the array
         * @since 1.8
         */
        public static <T> Stream<T> stream(T[] array) {
            return stream(array, 0, array.length);
        }
        // java.util.Arrays#stream(T[], int, int)
        /**
         * Returns a sequential {@link Stream} with the specified range of the
         * specified array as its source.
         *
         * @param <T> the type of the array elements
         * @param array the array, assumed to be unmodified during use
         * @param startInclusive the first index to cover, inclusive
         * @param endExclusive index immediately past the last index to cover
         * @return a {@code Stream} for the array range
         * @throws ArrayIndexOutOfBoundsException if {@code startInclusive} is
         *         negative, {@code endExclusive} is less than
         *         {@code startInclusive}, or {@code endExclusive} is greater than
         *         the array size
         * @since 1.8
         */
        public static <T> Stream<T> stream(T[] array, int startInclusive, int endExclusive) {
            // 構造 iterator, 帶入 StreamSupport 中
            return StreamSupport.stream(spliterator(array, startInclusive, endExclusive), false);
        }
    
        /**
         * Returns a {@link Spliterator} covering the specified range of the
         * specified array.
         *
         * <p>The spliterator reports {@link Spliterator#SIZED},
         * {@link Spliterator#SUBSIZED}, {@link Spliterator#ORDERED}, and
         * {@link Spliterator#IMMUTABLE}.
         *
         * @param <T> type of elements
         * @param array the array, assumed to be unmodified during use
         * @param startInclusive the first index to cover, inclusive
         * @param endExclusive index immediately past the last index to cover
         * @return a spliterator for the array elements
         * @throws ArrayIndexOutOfBoundsException if {@code startInclusive} is
         *         negative, {@code endExclusive} is less than
         *         {@code startInclusive}, or {@code endExclusive} is greater than
         *         the array size
         * @since 1.8
         */
        public static <T> Spliterator<T> spliterator(T[] array, int startInclusive, int endExclusive) {
            return Spliterators.spliterator(array, startInclusive, endExclusive,
                                            Spliterator.ORDERED | Spliterator.IMMUTABLE);
        }
        // java.util.stream.StreamSupport#stream(java.util.Spliterator<T>, boolean)
        /**
         * Creates a new sequential or parallel {@code Stream} from a
         * {@code Spliterator}.
         *
         * <p>The spliterator is only traversed, split, or queried for estimated
         * size after the terminal operation of the stream pipeline commences.
         *
         * <p>It is strongly recommended the spliterator report a characteristic of
         * {@code IMMUTABLE} or {@code CONCURRENT}, or be
         * <a href="../Spliterator.html#binding">late-binding</a>.  Otherwise,
         * {@link #stream(java.util.function.Supplier, int, boolean)} should be used
         * to reduce the scope of potential interference with the source.  See
         * <a href="package-summary.html#NonInterference">Non-Interference</a> for
         * more details.
         *
         * @param <T> the type of stream elements
         * @param spliterator a {@code Spliterator} describing the stream elements
         * @param parallel if {@code true} then the returned stream is a parallel
         *        stream; if {@code false} the returned stream is a sequential
         *        stream.
         * @return a new sequential or parallel {@code Stream}
         */
        public static <T> Stream<T> stream(Spliterator<T> spliterator, boolean parallel) {
            Objects.requireNonNull(spliterator);
            return new ReferencePipeline.Head<>(spliterator,
                                                StreamOpFlag.fromCharacteristics(spliterator),
                                                parallel);
        }
            // java.util.stream.ReferencePipeline.Head#Head(java.util.Spliterator<?>, int, boolean)
            /**
             * Constructor for the source stage of a Stream.
             *
             * @param source {@code Spliterator} describing the stream source
             * @param sourceFlags the source flags for the stream source, described
             *                    in {@link StreamOpFlag}
             */
            Head(Spliterator<?> source,
                 int sourceFlags, boolean parallel) {
                super(source, sourceFlags, parallel);
            }
        // java.util.stream.ReferencePipeline#ReferencePipeline(java.util.Spliterator<?>, int, boolean)
        /**
         * Constructor for the head of a stream pipeline.
         *
         * @param source {@code Spliterator} describing the stream source
         * @param sourceFlags The source flags for the stream source, described in
         *        {@link StreamOpFlag}
         * @param parallel {@code true} if the pipeline is parallel
         */
        ReferencePipeline(Spliterator<?> source,
                          int sourceFlags, boolean parallel) {
            super(source, sourceFlags, parallel);
        }
        // java.util.stream.AbstractPipeline#AbstractPipeline(java.util.Spliterator<?>, int, boolean)
        /**
         * Constructor for the head of a stream pipeline.
         *
         * @param source {@code Spliterator} describing the stream source
         * @param sourceFlags the source flags for the stream source, described in
         * {@link StreamOpFlag}
         * @param parallel {@code true} if the pipeline is parallel
         */
        AbstractPipeline(Spliterator<?> source,
                         int sourceFlags, boolean parallel) {
            this.previousStage = null;
            this.sourceSpliterator = source;
            this.sourceStage = this;
            this.sourceOrOpFlags = sourceFlags & StreamOpFlag.STREAM_MASK;
            // The following is an optimization of:
            // StreamOpFlag.combineOpFlags(sourceOrOpFlags, StreamOpFlag.INITIAL_OPS_VALUE);
            this.combinedFlags = (~(sourceOrOpFlags << 1)) & StreamOpFlag.INITIAL_OPS_VALUE;
            this.depth = 0;
            this.parallel = parallel;
        }
    

      如上,就返回了一 Stream 的具體實例,即是 ReferencePipeline.Head 的實例。故而,之后的每個stream操作如 filter,map,foreach方法,都盡在該 head 中進行實現了。一瞅便知。

        // java.util.stream.ReferencePipeline#filter
        @Override
        public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {
            Objects.requireNonNull(predicate);
            // 只返回了一個 StreamlessOp實例
            return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,
                                         StreamOpFlag.NOT_SIZED) {
                @Override
                Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
                    return new Sink.ChainedReference<P_OUT, P_OUT>(sink) {
                        @Override
                        public void begin(long size) {
                            downstream.begin(-1);
                        }
    
                        @Override
                        public void accept(P_OUT u) {
                            // 在必要時候調用 test() 方法即可
                            // 當test返回 true 時,該元素被保留傳入下一級調用中,此即filter的語義
                            if (predicate.test(u))
                                downstream.accept(u);
                        }
                    };
                }
            };
        }
        // java.util.stream.ReferencePipeline#map
        @Override
        @SuppressWarnings("unchecked")
        public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) {
            Objects.requireNonNull(mapper);
            // 同樣,僅返回一個 StatelessOp 的實例
            return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
                                         StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
                @Override
                Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
                    return new Sink.ChainedReference<P_OUT, R>(sink) {
                        @Override
                        public void accept(P_OUT u) {
                            // 同樣,在必要的時候調用 apply 方法
                            // 即 map 的語義為 每個元素都會調用該方法
                            downstream.accept(mapper.apply(u));
                        }
                    };
                }
            };
        }
        @Override
        public final <R> Stream<R> flatMap(Function<? super P_OUT, ? extends Stream<? extends R>> mapper) {
            Objects.requireNonNull(mapper);
            // We can do better than this, by polling cancellationRequested when stream is infinite
            return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
                                         StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {
                @Override
                Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
                    return new Sink.ChainedReference<P_OUT, R>(sink) {
                        @Override
                        public void begin(long size) {
                            downstream.begin(-1);
                        }
    
                        @Override
                        public void accept(P_OUT u) {
                            // flatmap 語義,所得結果,依次往下傳輸
                            try (Stream<? extends R> result = mapper.apply(u)) {
                                // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it
                                if (result != null)
                                    result.sequential().forEach(downstream);
                            }
                        }
                    };
                }
            };
        }

     

      如上,幾個方法調用下來,我們基本都可以看到,都是一個個的 StatelessOp 的實例的返回,但都沒有觸發真正的計算。那么,真正計算又要到幾時呢?相信有些其他知識面的你,定然會想到,在合適的時候再來觸發真正的運算操作。當數據結構不會發生本質的變化時,這種平衡就是存在的。只是在一些關鍵時候,才會觸發運算。這為后續進行并行計算或者性能優化提供了可能。

      那么,stream包中,哪些運算是作為真正的觸發行為呢?至少 collect(), foreach(), reduce() 是會進行觸發的。 這些優化手段,不知和其他框架實現,誰先誰后,誰主誰從。反正,總是好的想法。在其他地方,也許叫許多算子。

      我們以collect()探查如何使用這stream的威力?

        // java.util.stream.ReferencePipeline#collect(java.util.stream.Collector<? super P_OUT,A,R>)
        @Override
        @SuppressWarnings("unchecked")
        public final <R, A> R collect(Collector<? super P_OUT, A, R> collector) {
            A container;
            // 即分并行與串行
            if (isParallel()
                    && (collector.characteristics().contains(Collector.Characteristics.CONCURRENT))
                    && (!isOrdered() || collector.characteristics().contains(Collector.Characteristics.UNORDERED))) {
                container = collector.supplier().get();
                BiConsumer<A, ? super P_OUT> accumulator = collector.accumulator();
                forEach(u -> accumulator.accept(container, u));
            }
            else {
                // 串行執行
                container = evaluate(ReduceOps.makeRef(collector));
            }
            return collector.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)
                   ? (R) container
                   : collector.finisher().apply(container);
        }
    
        /**
         * Constructs a {@code TerminalOp} that implements a mutable reduce on
         * reference values.
         *
         * @param <T> the type of the input elements
         * @param <I> the type of the intermediate reduction result
         * @param collector a {@code Collector} defining the reduction
         * @return a {@code ReduceOp} implementing the reduction
         */
        public static <T, I> TerminalOp<T, I>
        makeRef(Collector<? super T, I, ?> collector) {
            Supplier<I> supplier = Objects.requireNonNull(collector).supplier();
            BiConsumer<I, ? super T> accumulator = collector.accumulator();
            BinaryOperator<I> combiner = collector.combiner();
            class ReducingSink extends Box<I>
                    implements AccumulatingSink<T, I, ReducingSink> {
                @Override
                public void begin(long size) {
                    state = supplier.get();
                }
    
                @Override
                public void accept(T t) {
                    accumulator.accept(state, t);
                }
    
                @Override
                public void combine(ReducingSink other) {
                    state = combiner.apply(state, other.state);
                }
            }
            // 返回ReuceOp
            return new ReduceOp<T, I, ReducingSink>(StreamShape.REFERENCE) {
                @Override
                public ReducingSink makeSink() {
                    return new ReducingSink();
                }
    
                @Override
                public int getOpFlags() {
                    return collector.characteristics().contains(Collector.Characteristics.UNORDERED)
                           ? StreamOpFlag.NOT_ORDERED
                           : 0;
                }
            };
        }
    
        // 運算一系列任務
        /**
         * Evaluate the pipeline with a terminal operation to produce a result.
         *
         * @param <R> the type of result
         * @param terminalOp the terminal operation to be applied to the pipeline.
         * @return the result
         */
        final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {
            assert getOutputShape() == terminalOp.inputShape();
            if (linkedOrConsumed)
                throw new IllegalStateException(MSG_STREAM_LINKED);
            linkedOrConsumed = true;
    
            return isParallel()
                   ? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))
                   : terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
        }
    
            // java.util.stream.ReduceOps.ReduceOp#evaluateSequential
            @Override
            public <P_IN> R evaluateSequential(PipelineHelper<T> helper,
                                               Spliterator<P_IN> spliterator) {
                return helper.wrapAndCopyInto(makeSink(), spliterator).get();
            }
    
        // java.util.stream.AbstractPipeline#wrapAndCopyInto
        @Override
        final <P_IN, S extends Sink<E_OUT>> S wrapAndCopyInto(S sink, Spliterator<P_IN> spliterator) {
            copyInto(wrapSink(Objects.requireNonNull(sink)), spliterator);
            return sink;
        }
    
        // java.util.stream.AbstractPipeline#wrapSink
        @Override
        @SuppressWarnings("unchecked")
        final <P_IN> Sink<P_IN> wrapSink(Sink<E_OUT> sink) {
            Objects.requireNonNull(sink);
            // 基本是按照倒序來排的
            for ( @SuppressWarnings("rawtypes") AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previousStage) {
                // 一層層包裝算子
                sink = p.opWrapSink(p.previousStage.combinedFlags, sink);
            }
            return (Sink<P_IN>) sink;
        }
    
        // java.util.stream.AbstractPipeline#copyInto
        @Override
        final <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
            Objects.requireNonNull(wrappedSink);
            // 依次調用 begin, foreach, end 方法
            if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
                wrappedSink.begin(spliterator.getExactSizeIfKnown());
                // 每個元素依次迭代, 一層層退出來
                spliterator.forEachRemaining(wrappedSink);
                wrappedSink.end();
            }
            else {
                copyIntoWithCancel(wrappedSink, spliterator);
            }
        }
    
            // java.util.Spliterators.ArraySpliterator#forEachRemaining
            @SuppressWarnings("unchecked")
            @Override
            public void forEachRemaining(Consumer<? super T> action) {
                Object[] a; int i, hi; // hoist accesses and checks from loop
                if (action == null)
                    throw new NullPointerException();
                if ((a = array).length >= (hi = fence) &&
                    (i = index) >= 0 && i < (index = hi)) {
                    do { action.accept((T)a[i]); } while (++i < hi);
                }
            }

      可見,該stream包的實現中,大量使用了包裝器模式,責任鏈模式,模板方法模式,以及在必要的節點再進行統一的運算觸發。且在必要的時候開啟并行計算,為上層應用帶了各種可能。在使用起來極其簡單的同時,又兼顧了性能。(我說的不是通常的性能,比如我自己寫幾個簡單的filter豈不性能更好?)而以上,僅僅是 stream 中的一種實現,針對每個不同類型的數據,其處理方式自然不一樣。比如 IntStream, DoubleStream, LongStream 雖同為Stream,但特性都都不一樣,不能一概而論。當然,一般這些實現都會遵守一定的接口規范。

      其中,以上這些簡便的寫法,得益于lamda語法的支持,以及幾個簡單的函數式接口定義。比如 Consumer, Function... 它們都被定義在java.util.function包下面。

    @FunctionalInterface
    public interface Consumer<T> {
    
        /**
         * Performs this operation on the given argument.
         *
         * @param t the input argument
         */
        void accept(T t);
    
        /**
         * Returns a composed {@code Consumer} that performs, in sequence, this
         * operation followed by the {@code after} operation. If performing either
         * operation throws an exception, it is relayed to the caller of the
         * composed operation.  If performing this operation throws an exception,
         * the {@code after} operation will not be performed.
         *
         * @param after the operation to perform after this operation
         * @return a composed {@code Consumer} that performs in sequence this
         * operation followed by the {@code after} operation
         * @throws NullPointerException if {@code after} is null
         */
        default Consumer<T> andThen(Consumer<? super T> after) {
            Objects.requireNonNull(after);
            return (T t) -> { accept(t); after.accept(t); };
        }
    }
    @FunctionalInterface
    public interface Function<T, R> {
    
        /**
         * Applies this function to the given argument.
         *
         * @param t the function argument
         * @return the function result
         */
        R apply(T t);
    
        /**
         * Returns a composed function that first applies the {@code before}
         * function to its input, and then applies this function to the result.
         * If evaluation of either function throws an exception, it is relayed to
         * the caller of the composed function.
         *
         * @param <V> the type of input to the {@code before} function, and to the
         *           composed function
         * @param before the function to apply before this function is applied
         * @return a composed function that first applies the {@code before}
         * function and then applies this function
         * @throws NullPointerException if before is null
         *
         * @see #andThen(Function)
         */
        default <V> Function<V, R> compose(Function<? super V, ? extends T> before) {
            Objects.requireNonNull(before);
            return (V v) -> apply(before.apply(v));
        }
    
        /**
         * Returns a composed function that first applies this function to
         * its input, and then applies the {@code after} function to the result.
         * If evaluation of either function throws an exception, it is relayed to
         * the caller of the composed function.
         *
         * @param <V> the type of output of the {@code after} function, and of the
         *           composed function
         * @param after the function to apply after this function is applied
         * @return a composed function that first applies this function and then
         * applies the {@code after} function
         * @throws NullPointerException if after is null
         *
         * @see #compose(Function)
         */
        default <V> Function<T, V> andThen(Function<? super R, ? extends V> after) {
            Objects.requireNonNull(after);
            return (T t) -> after.apply(apply(t));
        }
    
        /**
         * Returns a function that always returns its input argument.
         *
         * @param <T> the type of the input and output objects to the function
         * @return a function that always returns its input argument
         */
        static <T> Function<T, T> identity() {
            return t -> t;
        }
    }
    
    @FunctionalInterface
    public interface Supplier<T> {
    
        /**
         * Gets a result.
         *
         * @return a result
         */
        T get();
    }
    View Code

          話說為何單叫lamda式寫法又叫作函數式編程?想來原因有二,一是調用手法像是函數一般,只須傳入參數即可調用,二來lamda實現方式為生出靜態函數調用而成。不知是也不是。 

    posted @ 2021-06-12 22:49  等你歸去來  閱讀(530)  評論(3編輯  收藏  舉報
    国产美女a做受大片观看