Collector và Spliterator tùy chỉnh

1. Collector tùy chỉnh: khi nào và cách tự viết

Trong Java Stream API, giao diện Collector được dùng để biến đổi một stream thành một collection hoặc một giá trị tổng hợp. Thông thường bạn sử dụng các collector có sẵn từ lớp Collectors (toList(), toMap(), groupingBy() v.v.), nhưng đôi khi bạn cần thứ gì đó đặc biệt — và khi đó bạn có thể viết Collector của riêng mình.

Collector là một đối tượng mô tả cách thu thập các phần tử của stream để tạo ra kết quả cuối cùng. Nó xác định bốn (thực ra là năm) thành phần chính:

• supplier — tạo container mới để thu thập phần tử (ví dụ, một danh sách hoặc một map mới).
• accumulator — thêm phần tử tiếp theo vào container.
• combiner — hợp nhất hai container (quan trọng đối với stream song song!).
• finisher — chuyển container thành kết quả cuối cùng (ví dụ, biến nó thành bất biến hoặc chuyển sang kiểu khác).
• characteristics — tập các cờ mô tả thuộc tính của collector (ví dụ, có hỗ trợ song song không, có thay đổi kiểu kết quả hay không, v.v.).

Khai báo kiểu:

Collector<T, A, R>

• T — kiểu phần tử của stream,
• A — kiểu bộ tích lũy trung gian,
• R — kiểu kết quả.

2. Ví dụ: Collector cho MultiMap (Map<K, List<V>>)

Giả sử bạn muốn thu thập stream các cặp Pair<K, V> thành Map<K, List<V>> (multi-map), trong đó mỗi khóa tương ứng với một danh sách giá trị.

Ví dụ triển khai:

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
    );
}

Cách dùng:

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. Ví dụ: Collector cho top-N phần tử

Giả sử bạn muốn thu thập một stream thành danh sách gồm N phần tử lớn nhất (ví dụ, top-5 theo thứ tự giảm dần).

Triển khai:

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(); // loại bỏ phần tử nhỏ nhất
        },
        (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()); // theo thứ tự giảm dần
            return result;
        },
        Collector.Characteristics.UNORDERED
    );
}

Cách dùng:

List<Integer> top3 = Stream.of(5, 1, 9, 3, 7, 2).collect(topN(3, Comparator.naturalOrder()));
// top3: [9, 7, 5]

4. Khi KHÔNG nên viết Collector riêng

• Nếu có thể diễn đạt bài toán bằng cách kết hợp các collector tiêu chuẩn và các thao tác downstream (groupingBy, mapping, flatMapping, collectingAndThen v.v.), hãy ưu tiên dùng chúng.
• Collector riêng chỉ cần cho những kịch bản thực sự không chuẩn (cấu trúc dữ liệu đặc biệt, phép tổng hợp phức tạp, top-N, multi-map, v.v.).
• Đừng viết Collector chỉ để có Collector — điều đó làm phức tạp việc bảo trì và kiểm thử.

Ví dụ:

// Thay vì tự viết Collector cho Map<K, Set<V>>:
.collect(Collectors.groupingBy(
    Pair::key,
    Collectors.mapping(Pair::value, Collectors.toSet())
))

5. Spliterator tùy chỉnh: để làm gì và cách làm

Spliterator là một giao diện đặc biệt để duyệt và chia nhỏ các collection (hoặc nguồn dữ liệu khác) thành các phần, đặc biệt hữu ích cho xử lý song song. Khác với iterator thông thường, Spliterator có thể “tách” (split) collection thành các phần độc lập để xử lý song song.

Các phương thức chính:

• tryAdvance(Consumer<? super T> action) — xử lý phần tử tiếp theo.
• trySplit() — cố gắng chia collection thành hai phần (trả về một Spliterator mới cho một trong các phần).
• estimateSize() — ước lượng số phần tử còn lại.
• characteristics() — bitmask các đặc tính (ORDERED, SIZED, SUBSIZED v.v.).

trySplit: chiến lược chia tách

Chia tách cân bằng — quan trọng đối với stream song song: trySplit nên trả về các phần có kích thước xấp xỉ nhau để tải được phân bổ đồng đều.

Nếu không thể chia (ví dụ, quá ít phần tử) — trả về null.

Ví dụ: Spliterator để đọc tệp theo từng phần
Giả sử bạn có một tệp lớn và muốn xử lý theo từng đợt 1000 dòng, để không phải giữ tất cả trong bộ nhớ.

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() {
        // Đối với việc đọc luồng từ tệp, chia tách không có ý nghĩa — trả về null
        return null;
    }

    @Override
    public long estimateSize() {
        return Long.MAX_VALUE; // không biết trước
    }

    @Override
    public int characteristics() {
        return ORDERED | NONNULL;
    }
}

Cách dùng:

try (BufferedReader reader = Files.newBufferedReader(Path.of("big.txt"))) {
    StreamSupport.stream(new ChunkedLineSpliterator(reader, 1000), false)
        .forEach(chunk -> processChunk(chunk));
}

Các đặc tính của Spliterator

• ORDERED — các phần tử có thứ tự xác định (ví dụ, danh sách).
• SIZED — biết chính xác số lượng phần tử.
• SUBSIZED — mọi Spliterator nhận được qua trySplit cũng SIZED.
• IMMUTABLE — nguồn không thay đổi trong quá trình duyệt.
• CONCURRENT — nguồn hỗ trợ sửa đổi song song an toàn.
• DISTINCT, SORTED, NONNULL — các thuộc tính bổ sung.

Quan trọng: khai báo đúng các đặc tính — điều này ảnh hưởng tới tối ưu hóa của stream.

6. Ví dụ

• Đọc tệp theo từng phần (chunk) — cho phép xử lý tệp lớn theo từng phần, không tải toàn bộ vào bộ nhớ.
• Phân tích (parse) không cấp phát thừa — nếu bạn phân tích luồng byte/ký tự và muốn tối thiểu hóa đối tượng tạm, có thể hiện thực một Spliterator trả về “cửa sổ” hoặc “lát cắt” của mảng gốc.

Ví dụ: Spliterator để phân tích CSV theo từng dòng

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; // phân tích tuần tự
    }

    @Override
    public long estimateSize() {
        return Long.MAX_VALUE;
    }

    @Override
    public int characteristics() {
        return ORDERED | NONNULL;
    }
}

7. Tích hợp với parallel() — làm thế nào cho an toàn

• Nếu Spliterator của bạn hỗ trợ chia tách song song (trySplit không trả về null) và các đặc tính bao gồm SIZED/SUBSIZED, thì Stream API có thể song song hóa hiệu quả.
• Đối với nguồn dạng luồng (tệp, socket) thường không hỗ trợ chia tách — hãy dùng stream tuần tự.
• Đối với collection và mảng — hãy triển khai chia tách cân bằng (ví dụ, chia mảng làm đôi).

Ví dụ: Spliterator cho mảng

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;
    }
}

Cách dùng:

String[] arr = {"a", "b", "c", "d"};
StreamSupport.stream(new ArraySpliterator<>(arr, 0, arr.length), true)
    .forEach(System.out::println);