1. 独自 Collector: いつ、どのように作るか
Java の Stream API では、ストリームをコレクションや集約結果へ変換するのに Collector インターフェースを使います。通常は Collectors クラスの既製コレクタ（toList()、toMap()、groupingBy() など）を使いますが、特別な要件があるときには独自の Collector を作成できます。
Collector は、ストリームの要素から最終結果をどのように組み立てるかを記述するオブジェクトです。4（実際には5）つの主要コンポーネントを定義します:
- supplier — 要素を集めるための新しいコンテナ（例: 新しいリストやマップ）を生成します。
- accumulator — コンテナに次の要素を追加します。
- combiner — 2 つのコンテナを結合します（並列ストリームで重要）。
- finisher — コンテナを最終結果に変換します（不変化にする、別の型に変換するなど）。
- characteristics — コレクタの性質を表すフラグの集合（並列性をサポートするか、結果型を変更するか など）。
シグネチャ:
Collector<T, A, R>
- T — ストリーム要素の型
- A — 中間アキュムレータの型
- R — 結果型
2. 例: マルチマップ用 Collector (Map<K, List<V>>)
たとえば、Pair<K, V> のストリームを Map<K, List<V>>（マルチマップ）に収集し、各キーに値のリストを対応付けたいとします。
実装例:
public static <K, V> Collector<Pair<K, V>, ?, Map<K, List<V>>> toMultiMap() {
return Collector.of(
HashMap::new, // supplier
(map, pair) -> map.computeIfAbsent(pair.key(), k -> new ArrayList<>()).add(pair.value()), // accumulator
(map1, map2) -> { // combiner
map2.forEach((k, vList) -> map1.merge(k, vList, (l1, l2) -> { l1.addAll(l2); return l1; }));
return map1;
},
Function.identity(), // finisher
Collector.Characteristics.UNORDERED
);
}
使用例:
List<Pair<String, Integer>> pairs = List.of(
new Pair<>("a", 1), new Pair<>("b", 2), new Pair<>("a", 3)
);
Map<String, List<Integer>> multiMap = pairs.stream().collect(toMultiMap());
// multiMap: {a=[1, 3], b=[2]}
3. 例: Top-N 要素のための Collector
たとえば、ストリームを「N 個の最大要素」（例: 降順のトップ5）に収集したいとします。
実装:
public static <T> Collector<T, ?, List<T>> topN(int n, Comparator<? super T> comparator) {
return Collector.of(
() -> new PriorityQueue<>(n, comparator), // supplier
(pq, t) -> {
pq.offer(t);
if (pq.size() > n) pq.poll(); // 最小の要素を削除
},
(pq1, pq2) -> {
pq2.forEach(t -> {
pq1.offer(t);
if (pq1.size() > n) pq1.poll();
});
return pq1;
},
pq -> {
List<T> result = new ArrayList<>(pq);
result.sort(comparator.reversed()); // 降順
return result;
},
Collector.Characteristics.UNORDERED
);
}
使用例:
List<Integer> top3 = Stream.of(5, 1, 9, 3, 7, 2).collect(topN(3, Comparator.naturalOrder()));
// top3: [9, 7, 5]
4. 独自 Collector を書くべきではない場合
- 標準のコレクタと downstream 操作（groupingBy、mapping、flatMapping、collectingAndThen など）の組み合わせで表現できる場合は、それらを使うほうが良い。
- 独自 Collector が必要なのは、本当に非標準なシナリオ（特殊なデータ構造、複雑な集計、Top-N、マルチマップ等）のときだけ。
- Collector のための Collector を書くべきではありません。保守やテストを複雑にします。
例:
// Map<K, Set<V>> 用の自作 Collector の代わりに:
.collect(Collectors.groupingBy(
Pair::key,
Collectors.mapping(Pair::value, Collectors.toSet())
))
5. 独自 Spliterator: 目的と実装方法
Spliterator は、特に並列処理のために、コレクション（やその他のデータソース）を効率良く走査・分割するためのインターフェースです。通常のイテレータと異なり、Spliterator はコレクションを独立した部分に「分割（split）」して並列処理に回せます。
主要メソッド:
- tryAdvance(Consumer<? super T> action) — 次の要素を処理します。
- trySplit() — コレクションを 2 つに分割しようとします（片方の部分の Spliterator を返します）。
- estimateSize() — 残り要素数の推定。
- characteristics() — 特性のビットマスク（ORDERED、SIZED、SUBSIZED など）。
trySplit: 分割戦略
バランスのとれた分割 — 並列ストリームでは重要です。trySplit は、おおよそ同じ大きさの部分を返して、スレッドの負荷が均等になるようにすべきです。
分割できない場合（例: 要素が少ないなど）は、null を返します。
例: ファイルをチャンク単位で読む Spliterator
大きなファイルがあり、すべてをメモリに保持しないよう、1000 行ずつ（チャンク）処理したいとします。
public class ChunkedLineSpliterator implements Spliterator<List<String>> {
private final BufferedReader reader;
private final int chunkSize;
public ChunkedLineSpliterator(BufferedReader reader, int chunkSize) {
this.reader = reader;
this.chunkSize = chunkSize;
}
@Override
public boolean tryAdvance(Consumer<? super List<String>> action) {
List<String> chunk = new ArrayList<>(chunkSize);
try {
String line;
for (int i = 0; i < chunkSize && (line = reader.readLine()) != null; i++) {
chunk.add(line);
}
if (chunk.isEmpty()) return false;
action.accept(chunk);
return true;
} catch (IOException e) {
throw new UncheckedIOException(e);
}
}
@Override
public Spliterator<List<String>> trySplit() {
// ファイルをストリーミングで読む場合は分割に意味がないため、null を返す
return null;
}
@Override
public long estimateSize() {
return Long.MAX_VALUE; // 事前には不明
}
@Override
public int characteristics() {
return ORDERED | NONNULL;
}
}
使用例:
try (BufferedReader reader = Files.newBufferedReader(Path.of("big.txt"))) {
StreamSupport.stream(new ChunkedLineSpliterator(reader, 1000), false)
.forEach(chunk -> processChunk(chunk));
}
Spliterator の特性
- ORDERED — 要素が特定の順序で並ぶ（例: リスト）。
- SIZED — 正確な要素数が分かっている。
- SUBSIZED — trySplit で得られるすべての Spliterator も SIZED。
- IMMUTABLE — 走査中にソースが変化しない。
- CONCURRENT — ソースが安全な並行変更をサポートする。
- DISTINCT、SORTED、NONNULL — その他の性質。
重要: 特性を正しく指定しましょう。これはストリームの最適化に影響します。
6. 例
- ファイルをチャンク（小分け）で読む — 大きなファイルをメモリに全載せせずに、部分ごとに処理できます。
- 余分なアロケーションを避けるパース — バイト／文字ストリームをパースし、一時オブジェクトの生成を最小化したい場合、元配列の「ウィンドウ」や「スライス」を返す Spliterator を実装できます。
例: 1 行ごとに CSV をパースする Spliterator
public class CsvLineSpliterator implements Spliterator<String[]> {
private final BufferedReader reader;
public CsvLineSpliterator(BufferedReader reader) {
this.reader = reader;
}
@Override
public boolean tryAdvance(Consumer<? super String[]> action) {
try {
String line = reader.readLine();
if (line == null) return false;
action.accept(line.split(","));
return true;
} catch (IOException e) {
throw new UncheckedIOException(e);
}
}
@Override
public Spliterator<String[]> trySplit() {
return null; // 逐次パース
}
@Override
public long estimateSize() {
return Long.MAX_VALUE;
}
@Override
public int characteristics() {
return ORDERED | NONNULL;
}
}
7. parallel() との統合 — 安全に行うには
- Spliterator が並列分割をサポートしている（trySplit が null を返さない）うえ、特性に SIZED/SUBSIZED が含まれていれば、Stream API は効率的に並列化できます。
- ストリーミングソース（ファイル、ソケット）では通常分割はサポートされません — 逐次ストリームを使いましょう。
- コレクションや配列では、バランスのとれた分割を実装します（例: 配列を半分に分ける）。
例: 配列用 Spliterator
public class ArraySpliterator<T> implements Spliterator<T> {
private final T[] array;
private int start, end;
public ArraySpliterator(T[] array, int start, int end) {
this.array = array;
this.start = start;
this.end = end;
}
@Override
public boolean tryAdvance(Consumer<? super T> action) {
if (start < end) {
action.accept(array[start++]);
return true;
}
return false;
}
@Override
public Spliterator<T> trySplit() {
int mid = (start + end) >>> 1;
if (mid == start) return null;
ArraySpliterator<T> split = new ArraySpliterator<>(array, start, mid);
start = mid;
return split;
}
@Override
public long estimateSize() {
return end - start;
}
@Override
public int characteristics() {
return ORDERED | SIZED | SUBSIZED | IMMUTABLE;
}
}
使用例:
String[] arr = {"a", "b", "c", "d"};
StreamSupport.stream(new ArraySpliterator<>(arr, 0, arr.length), true)
.forEach(System.out::println);
GO TO FULL VERSION