小川のせせらぎもやがてはうねる奔流に　Stream インターフェース (1) - terminal operators part 1 -

OpenJDK にて JDK 8 の Feature Complete がリリースされたので、暇を見つけて Java 8 の新機能をあれこれイジっていこうかなと。　使用する JDK のバージョンは 8-ea-b90 です。

ってことで、まずは java.util.stream.Stream インターフェースに定義されているメソッドを見ていきまーす。　Stream は、Java 7 まででいう java.util.Iterator が一番近い型でしょうか。　一定の型のオブジェクトのシーケンス（連なり）をオブジェクト化したものです。　Iterator と違うところは filter() や map() のような、要素オブジェクトの取捨選択や変換を容易に行えるメソッド（多くが関数インターフェースを引数にとる）が豊富に定義されているところです。　また、しばしばそれらの処理が遅延して行われるため、メモリの効率化や無限シーケンスの扱いなども可能な場合があります・・・みたいな説明でいいかな？　詳しくはドキュメント参照のこと（ダウンロードが必要）。　

このシリーズの記事では、Stream インターフェースに定義されているメソッドを以下のように分類して見ていく予定。　Stream オブジェクトはいわゆる Map / Reduce 処理を行うよう設計されているので、それに従った分類になってます：

Stream 生成
Map 処理
- 中間演算子1 (intermadiate operators part 1)
- 中間演算子2 (intermadiate operators part2)
- 状態保有中間演算子 (stateful intermadiate operators (+ short circuiting stateful intermadiate operators))
Reduce 処理
- 終端演算子1 (terminal operators part 1)（今回）
- 終端演算子2 (terminal operators part 2)
- ショートカット終端演算子 (short-circuiting terminal operators)
プリミティブ型ストリーム

ちょっと小分けしすぎかな。　まぁ、いいでしょう。　Stream インターフェースは Scala のような関数型言語には大抵の場合似たようなものが定義されているほどありふれた型なので、使われている用語に適切な日本語訳があるかもしれませんが、拙者はそちら方面に素養がないもので適当な訳語を付けてます。　ご了承くださいませ。　ちょっと目次と順番が違ってますが、今回は terminal operators (part 1)。

Terminal Operator

terminal operator って、それっぽく訳すと「終端演算子」というのかな？　Stream オブジェクトは、filter() や map() のようなメソッドをチェーンのようにつなげて新たな Stream オブジェクトにどんどん変換していって、最後に何らかの別の型に変換する（もしくは値を返さずに副作用のみを行う）という使い方が基本ですが、terminal operator というのは、その「最後に何らかの別の型に変換する」という部分を担うメソッドのことです。

今回扱う terminal ooperator は以下の通り：

// 各要素に副作用のある処理を実行
void forEach(Consumer<? super T> action)
void forEachOrdered(Consumer<? super T> action)

// 要素数
long count()

// reduce
Optional<T> reduce(BinaryOperator<T> accumulator)
T reduce(T identity, BinaryOperator<T> accumulator)
<U> U reduce(U identity, BiFunction<U,? super T,U> accumulator, BinaryOperator<U> combiner)

// collect
<R> R collect(Collector<? super T,R> collector)
<R> R collect(Supplier<R> resultFactory, BiConsumer<R,? super T> accumulator, BiConsumer<R,R> combiner)

// 最小値・最大値
Optional<T> min(Comparator<? super T> comparator)
Optional<T> max(Comparator<? super T> comparator)

上記のうち forEach(), forEachOrdered() 以外のメソッドは forEach() を使って実装できますが、よく使う定型処理なので組み込みメソッドとして定義してあるんでしょう。　実際、関数型の言語には同名のメソッドが大抵定義されてたかと。

ちなみに、各メソッドの引数に Java 8 から導入された関数型インターフェース*1が多用されていますが、それぞれ

Consumer<T> ・・・ T オブジェクト1つを引数にとり、値を返さない (void) メソッド accept(T) : void を持つインターフェース。　必ず副作用を行う
BiConsumer<S, T> ・・・ S オブジェクト、T オブジェクトの2つを引数にとり、値を返さないメソッド accept(S, T) : void をもつインターフェース
BinaryOperartor<T> ・・・ T オブジェクト2つを引数にとり T オブジェクトを返すメソッド apply(T, T) : T を持つインターフェース。　二項演算子
BiFunction<S, T, U> ・・・S オブジェクト、T オブジェクトの2つを引数にとり、U オブジェクトを返すメソッド apply(S, T) : U を持つインターフェース。　2変数関数

という感じです。　詳しくは「こちら」参照。　Collector は java.util.stream パッケージ内のインターフェース。　関数型インターフェースではありません。

ではそれぞれの簡単なサンプルコードを見ていきましょう。

サンプルコード

forEach()

まずは一番基本の forEach() メソッド。　forEachOrdered() メソッドは要素の出現順が変更されないと保証されている以外は同じメソッドなので省略。　forEach() の返り値は void なので、このメソッドは副作用を起こすことが前提です。　それは引数が Consumer であることからもわかります。　とりあえず、Stream の要素を表示するサンプル：

Stream<String> stream = Stream.of("Java", "Groovy", "Scala", "Clojure");    // Stream オブジェクト生成
stream.forEach(System.out::println);

Java 8 から導入された、「::」演算子によるメソッド参照の取得なんかも使って見ました。　stream の要素（文字列 "Java", "Groovy" など）それぞれを引数にして System.out の println() メソッドを呼び出しています。

count()

次は Stream の要素を数えるメソッド。　Stream は通常、要素数を保持していないとかと思うので（実装知らないけど）、要素数を数えるのもコストがかかる覚悟が必要w　というか、Stream は要素のシーケンスを逆に辿れないので、要素数を数えるだけでもそれでお仕舞い。　使い方はまぁ、簡単：

Stream<String> stream = Stream.of("Java", "Groovy", "Scala", "Clojure");
assert stream.count() == 4L;    // count() は long 値を返す

返り値は long 値です。　別に 4L にしなくても今の場合は4で大丈夫だけど。　ちなみに count() は

stream.mapToLong(e -> 1L).sum();

と同じだそうです。

reduce()

reduce() メソッドは、Stream の各要素をある意味で「足し上げる」メソッドです。　3つのオーバーロードされたシグニチャがありますが、引数が1つ、2つのものはそんなに難しくないと思います：

Optional<T> reduce(BinaryOperator<T> accumulator)
T reduce(T identity, BinaryOperator<T> accumulator)

引数の BinaryOperator は2つの要素が与えられたときに、それらをどう「足す」か？を指定します。　2つ目の reduce() はそれに加えて初期値を指定します。　ちょっと注意が必要なのは、1つ目の reduce() は返り値が Optional 型だというところです。　Optional は値がない（空）かも知れない場合に使う型で、Stream が空の場合に空の Optional が返されます。　2つ目の reduce() は Stream が空でも初期値が与えられているので値が空になることがないため、Optional が使われていません。

引数が2つのものは for文を使って

// T reduce(T identity, BinaryOperator<T> accumulator)

T result = identity;
for (T element : this stream)
    result = accumulator.apply(result, element)
return result;

と書くのと同じだそうです。

引数が3つの reduce() は・・・シグニチャを見てもなんのことだかw　JavaDoc には以下のコードと同じだと書かれてます：

// <U> U reduce(U identity, BiFunction<U,? super T,U> accumulator, BinaryOperator<U> combiner)

U result = identity;
for (T element : this stream)
    result = accumulator.apply(result, element)
return result;

んー、引数が2つの reduce() とほとんど同じですな。　ちょっと返り値の型が変わってるだけ。　えっ、第3引数の combiner 使ってないじゃん・・・　どうもこれは combiner を使わなくても同じ結果を返すように実装を行わないといけないということのようで。　例えば

1 + 2 + 3 + 4 + 5

を計算する場合、第2引数の accumulator だけを使うと

((((1 + 2) + 3) + 4) + 5)

というふうに計算しますが、これを

(((1 + 2) + 3) + (4 + 5))

のように計算しても同じ結果になるようにしましょう、ということでしょう（結合法則を満たす）。　ここで、((1 + 2) + 3) と (4 + 5) の計算は第2引数の accumulator によって計算されますが、それらの結果を足す際に combiner を使うんだと思います。　まぁ、とりあえず動くサンプル書ければいいんじゃね？ってことでサンプルコード：

import java.util.Optional;

Stream<String> stream = Stream.of("Java", "Groovy", "Scala", "Clojure");

// 値を「足し上げる」演算のみを指定。　返り値は Optional
assert stream.reduce(String::concat) .equals( Optional.of("JavaGroovyScalaClojure") );

// 初期値と「足し上げる」演算を指定。　返り値は Stream の要素の型と同じ
assert stream.reduce("", String::concat) .equals( "JavaGroovyScalaClojure" );

// 要素を変換して「足し上げる」
assert stream.reduce(0, (i, s) -> i + s.length(), (i1, i2) -> i1 + i2) == 22;

最後のものは mapToInt() と IntStream#sum() 使って書いた方が簡単ですけどね。

collect()

collect() は reduce() と結構似てますが、「mutable reduction」とドキュメントに書いてあるとおり、（StringBuilder のような）Builder やコレクションのように可変なオブジェクトに対して Stream の各要素を「足し上げて」いくメソッドです。　「足し上げ」メソッドは reduce() のように結果値を返さなくてよく、collect() の返り値は、「足し上げ」られた Builder やコレクションになります。　引数が1つの collect() は java.util.stream.Collection オブジェクトを引数としてとりますが、Collection インターフェースについてはそのうち機会があれば。　今は java.util.stream.Collectors クラスからビルドインの Collector を使えるというだけで OK です。　ではサンプルコード：

Stream<String> stream = Stream.of("Java", "Groovy", "Scala", "Clojure");

// Collector を1つとる collect
assert stream.collect(Collectors.toStringJoiner(", ")).toString() .equals( "Java, Groovy, Scala, Clojure" );

// 上記 reduce() の3つ目に対応するもの
assert stream.collect(StringBuilder::new,
                      StringBuilder::append,
                      StringBuilder::append).toString() .equals( "JavaGroovyScalaClojure" );

// 2つ目のものをメソッド参照を使わずに書き換えたもの
assert stream.collect(() -> new StringBuilder(),
                      (sb, s) -> sb.append(s),
                      (sb1, sb2) -> sb1.append(sb2)).toString() .equals( "JavaGroovyScalaClojure" );

2つ目の collect() では第2引数と第3引数に StringBuilder::append を渡してますが、実際に使われているのはオーバーロードされた別の append() です。　3つ目の collect() では、それを明示的（でもないか）に示すためにラムダ式を使って書き換えてみました。　うーむ、Collector が関数インターフェースじゃないので使い方があんまり便利な感じがしないなぁ*2。

min() / max()

最後は最小値、最大値を取得するメソッド。　引数に大小関係を比較するための Comparator を指定する必要があります：

import java.util.Optional;

Stream<String> stream = Stream.of("Java", "Groovy", "Scala", "Clojure");

assert stream.min(String.CASE_INSENSITIVE_ORDER) .equals( Optional.of("Clojure") );    // 辞書順序
assert stream.max((s1, s2) -> s1.length() - s2.length()) .equals( Optional.of("Clojure") );    // 文字列長