Bài viết này không chỉ cung cấp giải thích về lý thuyết mà còn minh họa qua các ví dụ mã thực tế cách ứng dụng Stream API để giải quyết các vấn đề lập trình phổ biến.

Trong quá trình phát triển hàng ngày, có nhiều tình huống như chuyển đổi đối tượng, loại bỏ trùng lặp danh sách liên kết, gọi dịch vụ theo từng批次, v.v. Việc triển khai các tình huống này bằng vòng lặp for hoặc câu lệnh if-else thường khiến mã nguồn trở nên dài dòng, dễ xảy ra lỗi và kém hiệu quả. Qua quá trình xem xét mã nguồn dự án và sự hướng dẫn của các đồng nghiệp, đã học được nhiều cách sử dụng mới. Vì vậy, bài viết này sẽ tổng hợp về công nghệ Stream.

Giới thiệu ví dụ

Trong Java, khi cần thao tác với các phần tử trong mảng, Collection và các lớp tập hợp khác, thông thường chúng ta sẽ sử dụng vòng lặp để xử lý từng phần tử, hoặc sử dụng Stream để xử lý.

Giả sử chúng ta có yêu cầu sau: Từ một câu cho trước, trả về danh sách các từ có độ dài lớn hơn 5, sắp xếp theo thứ tự giảm dần độ dài và chỉ trả về tối đa 3 từ.

Trường học hoặc khi chưa tiếp cận Stream, chúng ta có thể viết hàm như sau:

public List<String> layTuDaiHonKyTu(@NotNull String cau) {
    // Tách câu để lấy thông tin các từ cụ thể
    String[] tu = cau.split(" ");
    List<String> danhSachTu = new ArrayList<>();
    // Vòng lặp kiểm tra độ dài từ, lọc ra các từ đáp ứng yêu cầu
    for (String tuHienTai : tu) {
        if (tuHienTai.length() > 5) {
            danhSachTu.add(tuHienTai);
        }
    }
    // Sắp xếp danh sách đã lọc theo độ dài
    danhSachTu.sort((o1, o2) -> o2.length() - o1.length());
    // Kiểm tra độ dài danh sách kết quả, nếu lớn hơn 3 thì cắt danh sách con 3 phần tử đầu tiên
    if (danhSachTu.size() > 3) {
        danhSachTu = danhSachTu.subList(0, 3);
    }
    return danhSachTu;
}

Tuy nhiên, nếu sử dụng Stream:

public List<String> layTuDaiHonKyTuBangStream(@NotNull String cau) {
    return Arrays.stream(cau.split(" "))
            .filter(tu -> tu.length() > 5)
            .sorted((o1, o2) -> o2.length() - o1.length())
            .limit(3)
            .collect(Collectors.toList());
}

Có thể thấy rõ ràng mã nguồn đã ngắn gọn gần một nửa. Vì vậy, hiểu và sử dụng tốt Stream giúp mã nguồn của chúng ta ngắn gọn, dễ hiểu và mạch lạc.

I. Giới thiệu cơ bản về Stream

Nói chung, có thể chia các thao tác Stream thành 3 loại:

• Tạo Stream
• Xử lý trung gian Stream
• Kết thúc Stream

Mỗi thao tác ống dẫn Stream chứa một số phương pháp, trước hết hãy liệt kê các API:

1.1 Bắt đầu ống dẫn

Chủ yếu chịu trách nhiệm tạo một Stream mới, hoặc dựa trên các đối tượng hiện có như mảng, List, Set, Map, v.v. để tạo Stream mới.

1.2 Ống dẫn trung gian

Chịu trách nhiệm xử lý Stream và trả về một đối tượng Stream mới, các thao tác ống dẫn trung gian có thể được xếp lớp.

API	Chức năng
filter()	Lọc các phần tử theo điều kiện, trả về Stream mới.
map()	Chuyển đổi phần tử hiện có thành đối tượng loại khác, logic một-một, trả về Stream mới.
flatMap()	Chuyển đổi phần tử hiện có thành đối tượng loại khác, logic một-nhiều, tức là một phần tử có thể chuyển đổi thành một hoặc nhiều phần tử mới, trả về Stream mới.
limit()	Giữ lại số lượng phần tử chỉ định ở đầu tập hợp, trả về Stream mới.
skip()	Bỏ qua số lượng phần tử chỉ định ở đầu tập hợp, trả về Stream mới.
concat()	Kết hợp dữ liệu của hai Stream thành một Stream mới, trả về Stream mới.
distinct()	Loại bỏ trùng lặp tất cả phần tử trong Stream, trả về Stream mới.
sorted()	Sắp xếp tất cả phần tử trong Stream theo quy tắc chỉ định, trả về Stream mới.
peek()	Duyệt và xử lý từng phần tử trong Stream, trả về Stream đã xử lý.

1.3 Kết thúc ống dẫn

Như tên gọi, sau khi thực hiện thao tác kết thúc, Stream sẽ kết thúc, có thể thực hiện một số xử lý logic nhất định hoặc trả về một số kết quả dữ liệu sau khi xử lý.

API	Chức năng
count()	Trả về số lượng phần tử sau khi xử lý Stream.
max()	Trả về giá trị lớn nhất sau khi xử lý Stream.
min()	Trả về giá trị nhỏ nhất sau khi xử lý Stream.
findFirst()	Tìm thấy phần tử đầu tiên đáp ứng điều kiện thì dừng xử lý Stream.
findAny()	Tìm thấy bất kỳ phần tử nào đáp ứng điều kiện thì thoát khỏi xử lý Stream. Với Stream tuần tự, nó giống với findFirst, với Stream song song thì hiệu quả hơn, tìm thấy ở bất kỳ phân đoạn nào sẽ dừng các logic tính toán tiếp theo.
anyMatch()	Trả về giá trị boolean, tương tự isContains(), dùng để kiểm tra có phần tử nào đáp ứng điều kiện không.
allMatch()	Trả về giá trị boolean, dùng để kiểm tra tất cả phần tử đều đáp ứng điều kiện.
noneMatch()	Trả về giá trị boolean, dùng để kiểm tra không có phần tử nào đáp ứng điều kiện.
collect()	Chuyển đổi Stream thành loại chỉ định, thông qua Collectors để chỉ định.
toArray()	Chuyển đổi Stream thành mảng.
iterator()	Chuyển đổi Stream thành đối tượng Iterator.
foreach()	Không trả về giá trị, duyệt từng phần tử và thực hiện logic xử lý đã cho.

II. Sử dụng phương thức Stream

2.1 map và flatMap

Trong dự án, thường xuyên thấy và sử dụng map và flatMap, ví dụ như mã:

Điểm giống và khác nhau giữa chúng là gì?

Map và flatMap đều dùng để chuyển đổi phần tử hiện có thành các phần tử khác, điểm khác biệt nằm ở:

• map phải là một-một, tức là mỗi phần tử chỉ có thể chuyển đổi thành một phần tử mới;
• flatMap có thể là một-nhiều, tức là mỗi phần tử có thể chuyển đổi thành một hoặc nhiều phần tử mới;

Hai hình ảnh dưới đây minh họa sự khác biệt giữa chúng:

map:

flatMap:

Ví dụ map:

Có một danh sách ID chuỗi, bây giờ cần chuyển đổi thành danh sách đối tượng khác.

/**
 * map dùng để: đổi một thành một
 */
List<String> danhSachId = Arrays.asList("205", "105", "308", "469", "627", "193", "111");
// Sử dụng thao tác Stream
List<DonDatModel> ketQua = danhSachId.stream()
    .map(id -> {
        DonDatModel model = new DonDatModel();
        model.setMaNhomId(id);
        return model;
    })
    .collect(Collectors.toList());
System.out.println(ketQua);

Sau khi thực thi, sẽ thấy mỗi phần tử được chuyển đổi thành phần tử mới tương ứng, nhưng tổng số phần tử trước và sau là như nhau:

[
DonDatModel{maNhomId='205', hinhAnh='null', duongDan='null', urlKhac='null'},
DonDatModel{maNhomId='105', hinhAnh='null', duongDan='null', urlKhac='null'},
DonDatModel{maNhomId='308', hinhAnh='null', duongDan='null', urlKhac='null'}, 
DonDatModel{maNhomId='469', hinhAnh='null', duongDan='null', urlKhac='null'}, 
DonDatModel{maNhomId='627', hinhAnh='null', duongDan='null', urlKhac='null'},
DonDatModel{maNhomId='193', hinhAnh='null', duongDan='null', urlKhac='null'}, 
DonDatModel{maNhomId='111', hinhAnh='null', duongDan='null', urlKhac='null'}
]

Ví dụ flatMap:

Bây giờ có một danh sách câu, cần trích xuất từng từ trong mỗi câu để có được danh sách tất cả các từ:

List<String> danhSachCau = Arrays.asList("xin chao the gioi","Xin Chao Thong Tin Ban Dau");
// Sử dụng thao tác Stream
List<String> ketQua2 = danhSachCau.stream()
    .flatMap(cau -> Arrays.stream(cau.split(" ")))
    .collect(Collectors.toList());
System.out.println(ketQua2);

Kết quả:

[xin, chao, the, gioi, Xin, Chao, Thong, Tin, Ban, Dau]

Ở đây cần bổ sung rằng, khi thực hiện flatMap, thực chất là trước tiên xử lý từng phần tử và trả về một Stream mới, sau đó mở rộng và hợp nhất nhiều Stream thành một Stream hoàn toàn mới, như sau:

2.2 phương thức peek và foreach

Peek và foreach đều có thể dùng để duyệt qua từng phần tử và xử lý.

Nhưng theo phần giới thiệu trước, peek thuộc về phương thức trung gian, trong khi foreach thuộc về phương thức kết thúc. Điều này có nghĩa là peek chỉ có thể là một bước xử lý ở giữa đường ống và không thể thực thi trực tiếp để có kết quả, nó phải có thao tác kết thúc khác theo sau; trong khi foreach là phương thức kết thúc không trả về giá trị, có thể thực thi trực tiếp các thao tác liên quan.

public void kiemTraPeekVaForeach() {
    List<String> danhSachCau = Arrays.asList("xin chao the gioi","He Thong Kien Truc");
    // Điểm minh họa 1: chỉ sử dụng peek, cuối cùng sẽ không thực thi
    System.out.println("----trước peek----");
    danhSachCau.stream().peek(cau -> System.out.println(cau));
    System.out.println("----sau peek----");
    // Điểm minh họa 2: chỉ sử dụng foreach, cuối cùng sẽ thực thi
    System.out.println("----trước foreach----");
    danhSachCau.stream().forEach(cau -> System.out.println(cau));
    System.out.println("----sau foreach----");
    // Điểm minh họa 3: peek có thêm thao tác kết thúc, peek sẽ thực thi
    System.out.println("----trước peek và count----");
    danhSachCau.stream().peek(cau -> System.out.println(cau)).count();
    System.out.println("----sau peek và count----");
}

Kết quả cho thấy, khi gọi riêng peek, nó không được thực thi, nhưng khi thêm thao tác kết thúc sau peek, nó sẽ được thực thi, trong khi foreach có thể được thực thi trực tiếp:

----trước peek----
----sau peek----
----trước foreach----
xin chao the gioi
He Thong Kien Truc
----sau foreach----
----trước peek và count----
xin chao the gioi
He Thong Kien Truc
----sau peek và count----

2.3 filter, sorted, distinct, limit

Đây đều là các phương thức thao tác trung gian Stream phổ biến, ý nghĩa cụ thể của các phương thức có trong bảng trên. Khi sử dụng, có thể chọn một hoặc nhiều phương thức để kết hợp hoặc sử dụng nhiều phương thức giống nhau cùng lúc:

public void layNguoiDungMucTieu() {
    List<String> danhSachId = Arrays.asList("205","10","308","49","627","193","111", "193");
    // Sử dụng thao tác Stream
    List<DonDatModel> ketQua = danhSachId.stream()
            .filter(s -> s.length() > 2)
            .distinct()
            .map(Integer::valueOf)
            .sorted(Comparator.comparingInt(o -> o))
            .limit(3)
            .map(id -> new DonDatModel(id))
            .collect(Collectors.toList());
    System.out.println(ketQua);
}

Đoạn mã trên xử lý theo logic:

1. Sử dụng filter để loại bỏ dữ liệu không đáp ứng điều kiện
2. Thao tác distinct để loại bỏ trùng lặp
3. Sử dụng map để chuyển chuỗi thành số nguyên
4. Sử dụng sorted để sắp xếp theo thứ tự tăng dần
5. Sử dụng limit để lấy 3 phần tử đầu tiên
6. Lại sử dụng map để chuyển id thành đối tượng DonDatModel
7. Sử dụng collect để thu thập dữ liệu cuối cùng vào list

Kết quả:

[DonDatModel{id=111},  DonDatModel{id=193},  DonDatModel{id=205}]

2.4 Kết thúc Stream với kết quả đơn giản

Như đã giới thiệu trước, các phương thức kết thúc như count, max, min, findAny, findFirst, anyMatch, allMatch, noneMatch thuộc về loại kết quả đơn giản ở đây. Đơn giản ở đây là kết quả có dạng số, boolean hoặc đối tượng Optional, v.v.

public void kiemTraPhuongThucKetThucDonGian() {
    List<String> danhSachId = Arrays.asList("205", "10", "308", "49", "627", "193", "111", "193");
    // Thống kê số lượng phần tử còn lại sau xử lý Stream
    System.out.println(danhSachId.stream().filter(s -> s.length() > 2).count());
    // Kiểm tra có phần tử nào bằng 205 không
    System.out.println(danhSachId.stream().filter(s -> s.length() > 2).anyMatch("205"::equals));
    // Thao tác findFirst
    danhSachId.stream().filter(s -> s.length() > 2)
            .findFirst()
            .ifPresent(s -> System.out.println("findFirst:" + s));
}

Kết quả cuối cùng:

6
true
findFirst:205

Một khi một Stream đã thực hiện thao tác kết thúc, không thể thực hiện các thao tác khác trên Stream này nữa, nếu không sẽ báo lỗi, xem ví dụ dưới đây:

public void xuLyStreamSauKhiDong() {
    List<String> danhSachId = Arrays.asList("205", "10", "308", "49", "627", "193", "111", "193");
    Stream<String> stream = danhSachId.stream().filter(s -> s.length() > 2);
    // Thống kê số lượng phần tử còn lại sau xử lý Stream
    System.out.println(stream.count());
    System.out.println("-----dưới đây sẽ báo lỗi-----");
    // Kiểm tra có phần tử nào bằng 205 không
    try {
        System.out.println(stream.anyMatch("205"::equals));
    } catch (Exception e) {
        e.printStackTrace();
        System.out.println(e.toString());
    }
    System.out.println("-----trên đây sẽ báo lỗi-----");
}

Kết quả:

-----dưới đây sẽ báo lỗi-----
java.lang.IllegalStateException: stream has already been operated upon or closed
-----trên đây sẽ báo lỗi-----
java.lang.IllegalStateException: stream has already been operated upon or closed
  at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:229)
  at java.util.stream.ReferencePipeline.anyMatch(ReferencePipeline.java:516)
  at Solution_0908.main(Solution_0908.java:55)

Vì stream đã thực hiện phương thức kết thúc count(), nên khi thực hiện lại phương thức anyMatch trên stream, sẽ báo lỗi "stream has already been operated upon or closed", điểm này cần đặc biệt lưu ý khi sử dụng.

2.5 Phương thức kết thúc với kết quả thu thập

Vì Stream chủ yếu được sử dụng trong các tình huống xử lý dữ liệu tập hợp, ngoài các phương thức kết thúc đơn giản đã nêu ở trên, có nhiều tình huống hơn là cần lấy một đối tượng kết dạng tập hợp như List, Set hoặc HashMap, v.v.

Ở đây cần đến phương thức collect, nó có thể hỗ trợ tạo ra các loại dữ liệu kết quả sau:

1. Một tập hợp như List, Set hoặc HashMap, v.v.; 2. Đối tượng StringBuilder, hỗ trợ nối nhiều chuỗi và trả về kết quả đã nối;

2. Một đối tượng có thể ghi nhận số lượng hoặc tính tổng (tính toán thống kê theo lô);

2.5.1 Tạo đối tượng tập hợp

Có thể coi là场景 thường được sử dụng nhất của collect:

 List<DonDatModel> danhSachDonDat = Arrays.asList(new DonDatModel("11"),
            new DonDatModel("22"),
            new DonDatModel("33"));

// Thu thập thành list
List<DonDatModel> danhSachThuThap = danhSachDonDat
        .stream()
        .filter(don -> don.getMaNhomId().equals("11"))
        .collect(Collectors.toList());
System.out.println("danhSachThuThap:" + danhSachThuThap);

// Thu thập thành Set
Set<DonDatModel> tapHopThuThap = danhSachDonDat
        .stream()
        .filter(don -> don.getMaNhomId().equals("22"))
        .collect(Collectors.toSet());
System.out.println("tapHopThuThap:" + tapHopThuThap);

// Thu thập thành HashMap, key là id, value là đối tượng DonDatModel
Map<String, DonDatModel> banDoThuThap = danhSachDonDat
        .stream()
        .filter(don -> don.getMaNhomId().equals("33"))
        .collect(Collectors.toMap(DonDatModel::getMaNhomId, Function.identity(), (k1, k2) -> k2));
System.out.println("banDoThuThap:" + banDoThuThap);

Kết quả:

danhSachThuThap:[DonDatModel{maNhomId='11', hinhAnh='null', duongDan='null', urlKhac='null'}]
tapHopThuThap:[DonDatModel{maNhomId='22', hinhAnh='null', duongDan='null', urlKhac='null'}]
banDoThuThap:{33=DonDatModel{maNhomId='33', hinhAnh='null', duongDan='null', urlKhac='null'}}

2.5.2 Tạo chuỗi nối

Nối các giá trị trong List hoặc mảng vào một chuỗi và phân cách bằng dấu phẩy, tình huống này chắc hẳn không còn xa lạ với mọi người. Nếu dùng vòng lặp và StringBuilder để nối, còn phải xem xét xử lý dấu phẩy thừa ở cuối, rất phức tạp:

public void kiemTraNoiChuoi() {
    List<String> danhSachId = Arrays.asList("205", "10", "308", "49", "627", "193", "111", "193");
    StringBuilder builder = new StringBuilder();
    for (String id : danhSachId) {
        builder.append(id).append(',');
    }
    // Xóa dấu phẩy nối thừa ở cuối
    builder.deleteCharAt(builder.length() - 1);
    System.out.println("Sau khi nối: " + builder.toString());
}

Nhưng bây giờ có Stream, sử dụng collect có thể thực hiện một cách dễ dàng:

public void kiemTraThuThapNoiChuoi() {
    List<String> danhSachId = Arrays.asList("205", "10", "308", "49", "627", "193", "111", "193");
    String ketQuaNoi = danhSachId.stream().collect(Collectors.joining(","));
    System.out.println("Sau khi nối: " + ketQuaNoi);
}

Cả hai cách đều cho kết quả hoàn toàn giống nhau, nhưng cách dùng Stream tinh tế hơn:

Sau khi nối: 205,10,308,49,627,193,111,193

2.5.3 Tính toán toán học theo lô

Còn một tình huống, khi sử dụng thực tế có thể ít gặp hơn, là sử dụng collect để tạo thông tin tổng hợp dữ liệu số, cũng có thể tìm hiểu cách triển khai:

public void kiemTraTinhToanSo() {
    List<Integer> danhSachId = Arrays.asList(10, 20, 30, 40, 50);
    // Tính giá trị trung bình
    Double trungBinh = danhSachId.stream().collect(Collectors.averagingInt(value -> value));
    System.out.println("Giá trị trung bình: " + trungBinh);
    // Thông tin thống kê dữ liệu
    IntSummaryStatistics thongKe = danhSachId.stream().collect(Collectors.summarizingInt(value -> value));
    System.out.println("Thông tin thống kê: " + thongKe);
}

Trong ví dụ trên, sử dụng phương thức collect để thực hiện toán học trên các giá trị trong list, kết quả như sau:

Giá trị trung bình: 30.0
Tổng hợp: IntSummaryStatistics{count=5, sum=150, min=10, average=30.000000, max=50}

III. Stream song song

3.1 Cơ chế của parallelStream

Sử dụng Stream song song có thể tận dụng hiệu quả phần cứng đa CPU của máy tính, tăng tốc độ thực thi logic. Stream song song bằng cách chia toàn bộ Stream thành nhiều đoạn, sau đó thực hiện song song các logic xử lý trên các đoạn Stream, và cuối cùng tổng hợp kết quả thực thi của các đoạn Stream thành một Stream hoàn chỉnh.

Có thể thấy từ mã nguồn của parallelStream rằng Stream song song thực chất là chia nhỏ nhiệm vụ, cuối cùng chuyển nhiệm vụ đến "toàn cục" ForkJoinPool của JDK8. Trong Fork-Join, ví dụ một ForkJoinPool có 4 luồng, có một hàng đợi nhiệm vụ, các nhiệm vụ con được tách ra từ một nhiệm vụ lớn sẽ được gửi đến hàng đợi nhiệm vụ của pool, 4 luồng lấy nhiệm vụ từ hàng đợi để thực thi, luồng nào thực thi nhiệm vụ nhanh sẽ nhận được nhiều nhiệm vụ hơn, chỉ khi hàng đợi không có nhiệm vụ thì luồng mới rảnh, đây chính là cơ chế "ăn cắp công việc".

Có thể hiểu rõ hơn tư tưởng "chia để trị" này qua hình ảnh dưới đây:

3.2 Hạn chế và ràng buộc

1. Phép duyệt trong parallelStream() phải đảm bảo an toàn luồng;

Nhiều người khi đã quen với xử lý luồng, thường thay thế vòng lặp for bằng foreach() luồng, trên thực tế điều này không nhất thiết hợp lý. Nếu chỉ là vòng lặp đơn giản, thực sự không cần sử dụng xử lý luồng, vì luồng được đóng gói nhiều logic xử lý phức tạp ở tầng dưới, về hiệu năng không có ưu thế.

2. Trong parallelStream(), không nên sử dụng trực tiếp luồng mặc định;

ForkJoinPool poolTuThiet = new ForkJoinPool(n);
poolTuThiet.submit(
    () -> danhSachKhachHang.parallelStream().thaoTacCụThể

3. Khi sử dụng parallelStream(), nên tránh các thao tác tốn thời gian;

Nếu gặp các thao tác tốn thời gian, nhiều thao tác IO, hoặc có thao tác sleep luồng, cần tránh sử dụng Stream song song.

IV. Một số vấn đề có thể gặp khi sử dụng Stream

4.1 parallelStream và toàn bộ tiến trình java dùng chung ForkJoinPool

Nếu sử dụng trực tiếp parallelStream().foreach sẽ mặc định dùng ForkJoinPool toàn cục, điều này dẫn đến nhiều nơi trong chương trình dùng chung một pool luồng, bao gồm cả các thao tác gc liên quan, vì vậy khi hàng đợi nhiệm vụ đầy sẽ xảy ra tình trạng tắc nghẽn, dẫn đến mọi nơi sử dụng ForkJoinPool trong chương trình đều gặp vấn đề.

4.2 Dữ liệu ThreadLocal trống sau khi sử dụng parallelStream

Stream song song được tạo ra khi thực thi thực chất là do khung ForkJoin tạo ra nhiều luồng để thực thi song song. Do bản thân ThreadLocal không có tính kế thừa, luồng được tạo mới tự nhiên không thể lấy được dữ liệu ThreadLocal của luồng cha.

4.3 Chuyển sang map, không chú ý đến việc key trùng nhau

Khi sử dụng Stream để chuyển thành map, cần chú ý vấn đề key trùng nhau.

V. Tổng kết Stream

5.1 Ưu điểm

1. Mã nguồn ngắn gọn hơn. Phong cách lập trình khai báo, dễ thể hiện ý định logic của mã nguồn.

2. Giảm phụ thuộc giữa các logic. Một logic xử lý trung gian của Stream, không cần quan tâm đến nội dung ở thượng nguồn và hạ nguồn, chỉ cần thực hiện logic của bản thân theo quy ước.

3. Trong các tình huống Stream song song, hiệu năng cao hơn so với vòng lặp tuần tự từng phần tử.

4. Giao diện hàm, đặc tính thực thi trễ, các thao tác ống dẫn trung gian không thực thi ngay lập tức dù có nhiều bước, chỉ khi gặp thao tác kết thúc mới bắt đầu thực thi, có thể tránh một số thao tác không cần thiết ở giữa.

5.2 Nhược điểm

Khó debug.