2020了你還不會Java8新特性？（六）Stream原始碼剖析
阿新 • • 發佈：2020-01-07
Stream流原始碼詳解
節前小插曲
AutoCloseable介面： 通過一個例子 舉例自動關閉流的實現。
public interface BaseStream<T, S extends BaseStream<T, S>>
extends AutoCloseable{} // BaseStream 繼承了這個介面。 Stream繼承了Streampublic class AutoCloseableTest implements AutoCloseable { public void dosomething() { System.out.println(" do something "); } @Override public void close() throws Exception { System.out.println(" close invoked "); } public static void main(String[] args) throws Exception { try ( AutoCloseableTest autoCloseableTest = new AutoCloseableTest()){ autoCloseableTest.dosomething(); } } }
執行結果如下： 自動呼叫了關閉流的方法
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. jdk提供了平行的 特化的流。 * * <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. 為了執行計算，流會被執行到一個流管道當中。 一個流管道包含了： 一個源。（數字來的地方） 0個或多箇中間操作（將一個stream轉換成另外一個Stream）。 一個終止操作（會生成一個結果，或者是一個副作用（求和，遍歷））。 流是延遲的，只有當終止操作被髮起的時候，才會執行中間操作。 * <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. 集合和流雖然有一些相似性，但是他們的差異是不同的。 集合是為了高效對於元素的管理和訪問。流並不會提供方式去直接操作流裡的元素。（集合關注的是資料的管理，流關注的是元素內容的計算） 如果流操作並沒有提供我們需要的功能，那麼我們可以使用傳統的iterator or spliterator去執行操作。 * <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>. 無一例外的。這種引數總是函式式介面的形式。也就是lambda表示式。除非特別指定，這些引數必須是非空的。 * <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.) 流擁有一個closed方法，實現了AutoCloseable，在他的父類裡。 最上面以舉例實現。 但是一個流 除了是I/O流（因為持有控制代碼等資源）才需要被關閉外，是不需要被關閉的。 大多數的流底層是集合、陣列或者是生成器函式。 他們並不需要特別的資源管理。如果需要被關閉，可以用try（）操作。 * <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. 流管道可以被序列或者並行操作。這種模式只是一個屬性而已。 初始化的時候會進行一個選擇。 比如說 stream() 是序列流。parallelStream()是並行流。 還可以通過sequential（）or parallel() 來進行修改。 以最後一個被呼叫的方法為準。 也可以用isParallel（）來進行查詢流是否是並行流。 * @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>> { // 具體舉例， 原始碼中有例子 Stream<T> filter(Predicate<? super T> predicate); // 過濾 <R> Stream<R> map(Function<? super T, ? extends R> mapper); //對映 IntStream mapToInt(ToIntFunction<? super T> mapper); LongStream mapToLong(ToLongFunction<? super T> mapper); DoubleStream mapToDouble(ToDoubleFunction<? super T> mapper); <R> Stream<R> flatMap(Function<? super T, ? extends Stream<? extends R>> mapper); //壓平 IntStream flatMapToInt(Function<? super T, ? extends IntStream> mapper); LongStream flatMapToLong(Function<? super T, ? extends LongStream> mapper); DoubleStream flatMapToDouble(Function<? super T, ? extends DoubleStream> mapper);、 Stream<T> distinct();// 去重 Stream<T> sorted(); //排序 Stream<T> sorted(Comparator<? super T> comparator); Stream<T> peek(Consumer<? super T> action); Stream<T> limit(long maxSize); // 截斷 void forEach(Consumer<? super T> action); // 遍歷 void forEachOrdered(Consumer<? super T> action); // 遍歷時執行操作 Object[] toArray(); // 轉陣列 T reduce(T identity, BinaryOperator<T> accumulator); // 匯聚， 返回一個匯聚的結果 <R> R collect(Supplier<R> supplier, BiConsumer<R, ? super T> accumulator, BiConsumer<R, R> combiner); // 收集器 。。。 }
自行參考父介面中的方法；
Stream中具體方法的詳解
分割迭代器：
/** * Returns a spliterator for the elements of this stream. * * <p>This is a <a href="package-summary.html#StreamOps">terminal * operation</a>. * * @return the element spliterator for this stream */ Spliterator<T> spliterator();
baseStream 原始碼講解
BaseStream 是所有流的父類 。
/**
* Base interface for streams, which are sequences of elements supporting
* sequential and parallel aggregate operations. The following example
* illustrates an aggregate operation using the stream types {@link Stream}
* and {@link IntStream}, computing the sum of the weights of the red widgets:
*
* <pre>{@code
* int sum = widgets.stream()
* .filter(w -> w.getColor() == RED)
* .mapToInt(w -> w.getWeight())
* .sum();
* }</pre>
*
* See the class documentation for {@link Stream} and the package documentation
* for <a href="package-summary.html">java.util.stream</a> for additional
* specification of streams, stream operations, stream pipelines, and
* parallelism, which governs the behavior of all stream types.
*
* @param <T> the type of the stream elements
* @param <S> the type of of the stream implementing {@code BaseStream}
* @since 1.8
* @see Stream
* @see IntStream
* @see LongStream
* @see DoubleStream
* @see <a href="package-summary.html">java.util.stream</a>
*/
public interface BaseStream<T, S extends BaseStream<T, S>> extends AutoCloseable
public interface Stream<T> extends BaseStream<T, Stream<T>>
BaseStream(){
Iterator<T> iterator(); 迭代器
Spliterator<T> spliterator(); 分割迭代器 。 這是一個流的終止操作。
boolean isParallel(); 是否是並行。
S sequential(); // 返回一個等價的序列流。 返回S是一個新的流物件
S parallel(); //返回一個並行流。
S unordered(); // 返回一個無序的流。
S onClose(Runnable closeHandler); //當前流.onClose、 當close呼叫時，呼叫此方法。
void close(); // 關閉流
}關閉處理器的舉例
/**
* Returns an equivalent stream with an additional close handler. Close
* handlers are run when the {@link #close()} method
* is called on the stream, and are executed in the order they were
* added. All close handlers are run, even if earlier close handlers throw
* exceptions. If any close handler throws an exception, the first
* exception thrown will be relayed to the caller of {@code close()}, with
* any remaining exceptions added to that exception as suppressed exceptions
* (unless one of the remaining exceptions is the same exception as the
* first exception, since an exception cannot suppress itself.) May
* return itself.
*
* <p>This is an <a href="package-summary.html#StreamOps">intermediate
* operation</a>.
*
* @param closeHandler A task to execute when the stream is closed
* @return a stream with a handler that is run if the stream is closed
*/
S onClose(Runnable closeHandler);
public static void main(String[] args) {
List<String> list = Arrays.asList("hello","world");
NullPointerException nullPointerException = new NullPointerException("myexception");
try (Stream<String> stream = list.stream()){
stream.onClose(()->{
System.out.println("aaa");
// throw new NullPointerException("first");
throw nullPointerException;
}).onClose(()->{
System.out.println("aaa");
throw nullPointerException;
}).forEach(System.out::println);
}
// 出現異常會被壓制，
// 如果是同一個異常物件，只會列印一次異常。 如果是多個異常物件。都會被列印。
}javadoc 中的介紹比任何資料都詳細。
Stream 原始碼分析。
stream();
/**
* Returns a sequential {@code Stream} with this collection as its source.
返回一個序列流，把這個集合當做源
* <p>This method should be overridden when the {@link #spliterator()}
* method cannot return a spliterator that is {@code IMMUTABLE},
* {@code CONCURRENT}, or <em>late-binding</em>. (See {@link #spliterator()}
* for details.)
當不能返回 三種方法 中的一個時，這個方法應該被重寫。
* @implSpec
* The default implementation creates a sequential {@code Stream} from the
* collection's {@code Spliterator}.
預設會從集合中建立一個序列流。 返回
* @return a sequential {@code Stream} over the elements in this collection
* @since 1.8
*/
default Stream<E> stream() {
return StreamSupport.stream(spliterator(), false);
}spliterator(); 分割迭代器
/**
* Creates a {@link Spliterator} over the elements in this collection.
*
* Implementations should document characteristic values reported by the
* spliterator. Such characteristic values are not required to be reported
* if the spliterator reports {@link Spliterator#SIZED} and this collection
* contains no elements.
* <p>The default implementation should be overridden by subclasses that
* can return a more efficient spliterator. In order to
* preserve expected laziness behavior for the {@link #stream()} and
* {@link #parallelStream()}} methods, spliterators should either have the
* characteristic of {@code IMMUTABLE} or {@code CONCURRENT}, or be
* <em><a href="Spliterator.html#binding">late-binding</a></em>.
預設的子類應該被重寫。為了保留parallelStream 和 stream的延遲行為。特性需要滿足IMMUTABLE 或者CONCURRENT
* If none of these is practical, the overriding class should describe the
* spliterator's documented policy of binding and structural interference,
* and should override the {@link #stream()} and {@link #parallelStream()}
* methods to create streams using a {@code Supplier} of the spliterator,
* as in:
* <pre>{@code
* Stream<E> s = StreamSupport.stream(() -> spliterator(), spliteratorCharacteristics)
* }</pre>
為什麼叫分割迭代器。先分割，在迭代。
如果不能滿足上述的要求，則重寫的時候應該滿足上述的需求、
* <p>These requirements ensure that streams produced by the
* {@link #stream()} and {@link #parallelStream()} methods will reflect the
* contents of the collection as of initiation of the terminal stream
* operation.
這些確保了流會返回的內容。
* @implSpec
* The default implementation creates a
* <em><a href="Spliterator.html#binding">late-binding</a></em> spliterator
* from the collections's {@code Iterator}. The spliterator inherits the
* <em>fail-fast</em> properties of the collection's iterator.
* <p>
* The created {@code Spliterator} reports {@link Spliterator#SIZED}.
預設會從集合的迭代器中創建出一個延遲的分割迭代器。 預設的迭代器 會有預設大小的迭代器。
* @implNote
* The created {@code Spliterator} additionally reports
* {@link Spliterator#SUBSIZED}.
*
* <p>If a spliterator covers no elements then the reporting of additional
* characteristic values, beyond that of {@code SIZED} and {@code SUBSIZED},
* does not aid clients to control, specialize or simplify computation.
* However, this does enable shared use of an immutable and empty
* spliterator instance (see {@link Spliterators#emptySpliterator()}) for
* empty collections, and enables clients to determine if such a spliterator
* covers no elements.
如果分割迭代器不包含任何元素。 其他的屬性對客戶端是沒有任何幫助的。 然而會促進分割迭代器共享的作用。
* @return a {@code Spliterator} over the elements in this collection
* @since 1.8
*/
@Override
default Spliterator<E> spliterator() {
return Spliterators.spliterator(this, 0);
}