Java ストリームAPI

Java 11 Gold 取得に向けた学習記録

# ストリームAPI

コレクションの各要素などを**ストリーム**という概念で扱うためのクラスが`java.util.stream`パッケージには用意されている。

https://docs.oracle.com/javase/jp/11/docs/api/java.base/java/util/stream/package-summary.html

ストリーム生成元のコレクションなどを**ソース**と表現する。基本的にはストリームを利用して各要素に操作を行ったとしても、ソースには影響が無い（[干渉](#干渉)、[副作用](#副作用)）。

ストリームを取得するには、いくつかの方法がある。

```java:リストからストリーム生成
List<String> list = List.of("E", "B", "D", "C", "A");
Stream<String> stream1 = list.stream();
```

```java:配列からストリーム生成
String[] array = new String[] {"E", "B", "D", "C", "A"};
Stream<String> stream2 = Arrays.stream(array);
```

# ストリームパイプライン

ストリームに対してメソッドチェーンを利用して複数の処理を実行する形式を**ストリームパイプライン**と言う。ストリームパイプラインは1つのソース、複数の**中間操作**、1つの**終端操作**から構成される。

メソッドチェーンは中間操作メソッドが実行元のインスタンスとは別の新たなストリームインスタンスを返すことによって実現している。

```java:ストリームのチェーンメソッド
Stream<String> stream = List.of("E", "B", "D", "C", "A").stream();

Stream<String> newStream = stream.sorted();

System.out.println(stream == newStream);
// >> false
```

中間操作と終端操作は`java.util.stream.Stream`インターフェースから提供される。

https://docs.oracle.com/javase/jp/11/docs/api/java.base/java/util/stream/Stream.html

## 中間操作

| 戻り値の型 | メソッド名 | 概要 |
|:-|:-|:-|
| `Stream<T>` | `distinct()` | 重複を排除する（`equals()`で判断） |
| `Stream<T>` | `filter(Predicate<? super T> predicate)` | `true`を返すもの要素だけに絞り込む |
| `Stream<T>` | `limit(long maxSize)` | 要素数を`maxSize`に制限 |
| `<R> Stream<R>` | `map(Function<? super T, ? extends R> mapper)` | 各要素に指定された関数を実行 |
| `Strema<T>` | `peek(Consumer<? super T> action)` | 新しいストリームを返すとともに、各要素に指定されたアクションを実行。ログ出力などやデバッグで利用。 |
| `Stream<T>` | `skip(long n)` | ストリームから最初のn個を破棄 |
| `Stream<T>` | `sorted(Comparator>? super T> comparator)` | `comparator`を使って並び替える |

## 終端操作

| 戻り値の型 | メソッド名 | 概要 |
|:-|:-|:-|
| `boolean` | `allMatch(Predicate<? super T> predicate)` | 全ての要素が`predicate`の結果が`true`になるかどうか |
| `boolean` | `anyMatch(Predicate<? super T> predicate)` | いずれかの要素が`predicate`の結果が`true`になるかどうか |
| `<R, A> R` | `collect(Collector<? super T,​A,​R> collector)` | `collector`を使って各要素に対する[可変リダクション](#可変リダクション)操作を実行 <br> コレクションを作るために使用する |
| `long` | `count()` | ストリームの要素数を返す |
| `Optional<T>` | `findAny()` | ストリーム内に要素が残っているかどうかの結果をラップした`Optional`を返す |
| `Optional<T>` | `findFirst()` | ストリーム内の最初の要素をラップした`Optional`を返す |
| `void` | `forEach(Consumer<? super T> action)` | 各要素に`action`を実行 |
| `Optional<T>` | `max(Comparator<? super T> omparator)` | `comparator`を使って最大の要素を返す |
| `Optional<T>` | `min(Comparator<? super T> omparator)` | `comparator`を使って最大の要素を返す |
| `boolean` | `noneMatch(Predicate<? super T> predicate)` | `predicate`が`true`を返す要素がないかどうか |
| `Optional<T>` | `reduce(BinaryOperator<T> accumulator)` | 各要素に[リダクション](#リダクション)操作を実行し、累積的に結合された結果を`Optional`でラップして返す（引数で初期値を指定した場合はラップされない） |
| `Object[]` | `toArray()` | ストリームを配列にする |
| `<A> A[]` | `toArray(IntFunction<A[]> generator)` | `generator`を使用してストリームを配列にする |


## `IllegalStateException`

`Stream`が継承する`BaseStraem`には`close()`メソッドが定義されていて、一度使用した`Stream`オブジェクトはこの`close()`メソッドによって閉じられることになっている。

もしも1度閉じられたストリームを再利用しようとした場合、`IllegalStateException`が`throw`される。

```java:IllegalStateException
Stream<String> usedStream = List.of("E", "B", "D", "C", "A").stream();

// この時点で usedStream は閉じられる
Stream<String> newStream1 = usedStream.sorted();

// 閉じられたストリームをもう一度使用した場合、IllegalStateException が発生する
Stream<String> newStream2 = usedStream.map((alphabet) -> {
    return "word:" + alphabet;
});
```

# 干渉
interference

ストリームパイプラインが元のソースに変更を加えることを**干渉**と言う。

基本的にストリームパイプラインにおいては干渉は避けるべきとされている。

```java:干渉
// ソース
List<String> list = List.of("E", "B", "D", "C", "A");

// ストリーム
Stream<String> stream = list.stream();

stream.map((alphabet) -> {
    // ソースに変更を加える処理（干渉）
    list.remove(alphabet);
    return alphabet;
});
```

# 順次・並列実行
sequential=順次
parallel=並列

ストリームパイプラインは順次、並列のいずれかで実行することができる。

例えば、`Collection`インターフェースの`stream()`で生成したストリームは**順次**実行となり、`parallelStream()`で生成したストリームは**並列**実行になる。

基本的には明示的に並列処理用のストリームを生成しない限りは、順次実行用のストリームになる。

```java:順次・並列
List<String> list = List.of("E", "B", "D", "C", "A");

// 順次実行のストリーム
Stream<String> stream = list.stream();

// 並列実行のストリーム
Stream<String> parallelStream = list.parallelStream();
```

# ステートレス操作・ステートフル操作

中間操作はさらに、**ステートレス操作**と**ステートフル操作**に分類される。

ステートレス操作は、各要素を処理するために他の要素の情報を必要としない。`map()`、`filter()`、`flatMap()`、`peek()`などが該当する。ステートレス操作は操作の結果がストリームの各要素に対して独立しているため、**並列処理に適している**。

ステートフル操作は、各要素を処理するために他の要素の情報を必要とする。`sorted()`、`distinct()`、`limit()`、`skip()`などが該当する。例えば`sorted()`では並び替えのために全体の要素数が確定した状態でしか実行できないし、`distinct()`は重複排除のために全体の要素を使用する必要がある。つまり、ステートフル操作は全体の要素の状態を把握する必要があり、各要素に対して独立して操作を実行することができないため、**並列処理に向かない**。

ステートレス操作とステートフル操作は実行のされ方も異なる。

ステートレス操作は各要素を独立して処理するため、ある要素の処理が他の要素の処理に影響を与えない。そのため一つの要素に対する処理が完了すると、次の要素への処理を待たず、処理が完了した要素を即座に次のステップに送信する。

一方、ステートフル操作は全要素を収集してから実行されるため、要素のうちの一つに対する処理が完了したとしても、即座に次のステップには進まず次の要素に対する処理を続けて実行する。

```java:ステートレス・ステートフル操作
List.of("E", "B", "D", "C", "A")
        .stream()
        .peek((alphabet) -> {
            System.out.println(alphabet);
            
        // ステートフル操作
        }).sorted((a, b) -> {
            System.out.println("sorting: " + a + ", " + b);
            return a.compareTo(b);
            
        // ステートレス操作
        }).peek((alphabet) -> {
            System.out.println("sorted: " + alphabet);
        }).map((alphabet) -> {
            System.out.println("mapping: " + alphabet);
            return "word:" + alphabet;
        }).peek((alphabet) -> {
            System.out.println("mapped: " + alphabet);
        }).forEach((alphabet) -> {
            System.out.println(alphabet);
        });

// >> E
// >> B
// >> D
// >> C
// >> A

// ステートフル操作

// >> sorting: B, E
// >> sorting: D, B
// >> sorting: D, E
// >> sorting: D, B
// >> sorting: C, D
// >> sorting: C, B
// >> sorting: A, D
// >> sorting: A, C
// >> sorting: A, B

// ステートレス操作

// >> sorted: A
// >> mapping: A
// >> mapped: word:A
// >> word:A
// >> sorted: B
// >> mapping: B
// >> mapped: word:B
// >> word:B
// >> sorted: C
// >> mapping: C
// >> mapped: word:C
// >> word:C
// >> sorted: D
// >> mapping: D
// >> mapped: word:D
// >> word:D
// >> sorted: E
// >> mapping: E
// >> mapped: word:E
// >> word:E
```

# 遅延実行

中間操作のためのメソッドの多くは関数型インターフェースを受け取るが、渡した「処理」（多くの場合、ラムダ式。もしくはメソッド参照）が内部で実行されるのは、終端操作の実行時になる。この特徴を**遅延実行**と言う。

```java:遅延実行
System.out.println("ストリームを生成");
Stream<String> stream = List.of("E", "B", "D", "C", "A").stream();

System.out.println("中間操作の実行");
Stream<String> stream2 = stream.sorted().map((alphabet) -> {
    System.out.println("map()に渡した処理の実行タイミングはここ");
    return "word:" + alphabet;
});

System.out.println("終端操作の実行");
String[] result = stream2.toArray(String[]::new);

for(String word : result){
    System.out.println(word);
}
// >> ストリームを生成
// >> 中間操作の実行
// >> 終端操作の実行
// >> map()に渡した処理の実行タイミングはここ
// >> map()に渡した処理の実行タイミングはここ
// >> map()に渡した処理の実行タイミングはここ
// >> map()に渡した処理の実行タイミングはここ
// >> map()に渡した処理の実行タイミングはここ
// >> word:A
// >> word:B
// >> word:C
// >> word:D
// >> word:E
```

実行結果から、中間操作が終端操作の開始をきっかけに開始されていることがわかる。

:::note info
中間操作は終端操作が呼び出された時点で開始される
:::


# 副作用

ストリームパイプラインが処理過程でパイプライン外部の状態を変更したり、外部と相互作用することを**副作用**と言う。

副作用は関数型プログラミングにおける純粋な関数が持つべき「副作用を持たない」と言う特徴に反するものであり、ストリームが本来のデータ処理の目的以外に影響を及ぼすことになる。またスレッドセーフにおける安全性に悪影響を与えることもある。

またストリームAPIには[遅延実行](#遅延実行)という特徴があることからも、副作用を持つ処理は避けるべきとされる。

```java:副作用がある
List<String> list = List.of("E", "B", "D", "C", "A");
Stream<String> stream = list.stream();

stream.map((alphabet) -> {
    return list.add("word:" + alphabet);
});
```

```java:副作用がない
List<String> list = List.of("E", "B", "D", "C", "A");
Stream<String> stream = list.stream();

List<String> result = stream.map((alphabet) -> {
    return "ward:" + alphabet;
}).collect(Collectors.toList());
```

副作用は[リダクション](#リダクション)を利用することでも回避できる場合がある。

# リダクション
Reduction

`reduce()`に代表される、ストリームの各要素を集約して単一の結果にまとめるような操作を**リダクション操作**と言う。

リダクション操作は元のソースに対して変更を加えないため、[干渉](#干渉)を避けることができる。

## `reduce()`

> `Optional<T> reduce​(BinaryOperator<T> accumulator)`

```java:reduce()
List<String> list = List.of("E", "B", "D", "C", "A");
Optional<String> result = list.stream().reduce( (sum, element) -> sum + element);
```

初期値（`identity`）を指定した場合、戻り値は初期値と同じ型になる。

> `T reduce​(T identity, BinaryOperator<T> accumulator)`

```java:reduce()
List<String> list = List.of("E", "B", "D", "C", "A");
String result = list.stream().reduce("", (sum, element) -> sum + element);
```

上の例ではメソッド参照を利用できる

```java
String result = list.stream().reduce("", String::concat);
```

## 可変リダクション
Mutable Reduction

`collect()`に代表される、ストリームの内部に存在する`Collection`や`StringBuilder`などの**可変結果コンテナ**がストリームの要素を逐次的に収集しながらも、途中でそのコンテナの中身を変更（更新）することができ、最終的には目的とする表現に変換することができるリダクション操作を**可変リダクション操作**と言う。

ここで、可変結果コンテナとは、ストリームの要素を収集するために使用される`List`、`Set`、`StringBuilder`などの変更可能なオブジェクトのことを指す。収集の過程でこのオブジェクトには新たな要素が追加され、コンテナとしての更新が行われている。

### `collect()`

`Collection`や`StringBuilder`などの可変結果コンテナ内に必要な要素を集める（collect）するところから`collect()`と言う名前になっている。

`collect()`を理解するには、`Collector`インターフェース、`Collectors`クラスについても理解する必要がある。

> `<R, A> R collect(Collector<? super T, A, R> collector)`

> `<R> R collect(Supplier<R> supplier, BiConsumer<R, ? super T> accumulator, BiConsumer<R, R> combiner)`

### `Collector`インターフェース

（個人的には最初、`Collector`は関数型インタフェースだと勘違いしていた。ストリームパイプラインの中間操作メソッドの多くは関数型インターフェースの実装クラスを引数に取るが、`collect()`メソッドは関数型インターフェースの実装ではなく、普通のインターフェース`Collector`の実装クラスを引数に受けとる。）

`java.util.stream.Collector`インターフェースは、**リダクション操作を定義する**ために使用される。

https://docs.oracle.com/javase/jp/11/docs/api/java.base/java/util/stream/Collector.html

`Collector`インターフェースは、`collect()`の引数で利用されている。

> `<R,​A> R collect​(Collector<? super T,​A,​R> collector)`

`Collector`インターフェース自体の型引数（ジェネリクス）は次のような定義になっている。

> `public interface Collector<T, A, R>`

`T` : 入力要素の型
`A` : 中間操作の`accumulator`の型（関数型インターフェース）
`R` : 集約結果の型

`Collector`インターフェースで定義された抽象メソッドは、主に次の4つ。これらは関数型インターフェースの実装を返すメソッドで、関数型インターフェースの実装部分がリダクション操作に当たる。

- `Supplier<A> supplier()`
- `BiConsumer<A, T> accumulator()`
- `BinaryOperator<A> combiner()`
- `Function<A, R> finisher()`

#### `Supplier<A> supplier()`

「新しい結果コンテナを作成する`Supplier`インターフェースの実装」を返す。

#### `BiConsumer<A, T> accumulator()`

「結果コンテナへ新しいデータ要素を追加する`BiConsumer`インターフェースの実装」を返す。

#### `BinaryOperator<A> combiner()`

「2つの結果コンテナを1つに結合する`BinaryOperator`インターフェースの実装」を返す。

#### `Function<A, R> finisher()`

(オプション)「コンテナに対する最終的な変換を行う`Function`インターフェースの実装」を返す。


#### 関数型インターフェースの実装を返す関数

`Collector`インターフェースに定義された各メソッドは、次のように関数型インターフェースの実装、すなわちラムダ式もしくはメソッド参照を返す。

> `interface Supplier<T> { T get(); }`

```java:関数を返すメソッド
Supplier<String> getSupplier() {
    return () -> {
        return "hello world";
    };
}
```

### `Collectors`クラス

`java.util.stream.Collectors`クラスには、前述の`Collector`インターフェースの実装を取得するためのファクトリメソッドが定義されている。

https://docs.oracle.com/javase/jp/11/docs/api/java.base/java/util/stream/Collectors.html

https://qiita.com/AsahinaKei/items/3a1f17b728769bab95b7

`Collectors`クラスのメソッドを使用することで、自前で`Collector`インターフェースを実装する手間を省くことができる。

### `collect()`、`Collector`、`Collectors`の関係性

`collect()`メソッドは、ストリーム要素を収集し、可変結果コンテナに格納する。

`Collector`インターフェースは、リダクション操作を取得するための関数を提供する。

`Collectors`クラスは、一般的なリダクション操作の実装を提供する。

![collect.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1069520/b4371450-aa5e-4197-24fd-d6ce1f50b227.png)

`collect()`メソッドにはオーバーロードが2つあり、`Collectors`クラスが提供する`Collector`インターフェースの実装オブジェクトを渡すもの以外にも、自分で実装を指定する方法も用意されている。

用意されたものを使う場合は、

> `<R,​A> R collect​(Collector<? super T,​A,​R> collector)`

を利用し、自前で実装する場合は

> `<R> R collect​(Supplier<R> supplier, BiConsumer<R,​? super T> accumulator, BiConsumer<R,​R> combiner)`

を利用する。

### `collect()`を使用する

```java:collect()を使用する
List<String> list = List.of("E", "B", "D", "C", "A");

Stream<String> stream = list.stream();

// リストをセットに変換
Set<String> result = stream.collect(Collectors.toSet());
```

# 参考

・<a href="https://amzn.asia/d/gkx4D5q" target="_blank" rel="noopener noreferrer">徹底攻略Java SE 11 Gold問題集[1Z0-816]対応</a>
・<a href="https://www.slideshare.net/slideshow/java8-launch/32567234" target="_blank" rel="noopener noreferrer">徹底解説！Project Lambdaのすべて リターンズ</a>
<a></a>