Bằng cách tận dụng sức mạnh của các biểu thức lambda, java.util.stream package giúp dễ dàng chạy các functional-style queries trên các collections, arrays và các tập dữ liệu khác.
Tính năng mới chính trong Java SE 8 là các biểu thức lambda. chúng ta có thể coi biểu thức lambda là một anonymous method(phương thức ẩn danh); giống như các method, lambdas có các typed parameters(tham số đã nhập), body(phần thân) và return type(kiểu trả về). Nhưng thứ thực sự mới ở đây không phải là bản thân các biểu thức lambda, mà là những gì chúng có khả năng làm được. Lambdas giúp dễ dàng thể hiện hành vi dưới dạng dữ liệu, từ đó giúp phát triển các thư viện mạnh mẽ hơn và ấn tượng hơn.

Một trong những thư viện như vậy, java.util.stream package cho phép biểu thức ngắn gọn và có tính khai báo của các hoạt động hàng loạt có thể song song trên nhiều nguồn dữ liệu khác nhau. Các thư viện như Streams có thể đã được viết trong các phiên bản Java trước đó, nhưng nếu không có quy ước về behavior-as-data(hành vi dưới dạng dữ liệu - Lambdas) nhỏ gọn thì chúng sẽ trở nên cồng kềnh đến mức không ai muốn sử dụng chúng. chúng ta có thể coi Streams là thư viện đầu tiên tận dụng sức mạnh của các biểu thức lambda trong Java, nhưng không có gì kỳ diệu về nó (mặc dù nó được tích hợp chặt chẽ vào các thư viện JDK cốt lõi). Streams không phải là một phần của ngôn ngữ — đó là một thư viện được thiết kế cẩn thận nhằm tận dụng một số tính năng ngôn ngữ mới hơn.

Bài viết này là bài đầu tiên trong loạt bài tìm hiểu sâu về thư viện java.util.stream. Phần này giới thiệu với chúng ta về thư viện và cung cấp cho chúng ta cái nhìn tổng quan về các ưu điểm và nguyên tắc thiết kế của nó. Trong các phần tiếp theo, chúng ta sẽ tìm hiểu cách sử dụng các luồng để tổng hợp và xử lý dữ liệu, đồng thời xem qua nội bộ của thư viện và tối ưu hóa hiệu suất.

Querying with streams

Một trong những cách sử dụng phổ biến nhất của luồng là biểu thị các truy vấn đối với data trong các collections. Ví dụ 1 cho thấy một ví dụ về một đường dẫn truyền phát đơn giản. Quy trình này lấy một tập hợp các giao dịch mua hàng giữa người mua và người bán, đồng thời tính toán tổng giá trị đồng đô la của các giao dịch của người bán sống ở New York.

Ví dụ 1

int totalSalesFromNY
    = txns.stream()
          .filter(t ‑> t.getSeller().getAddr().getState().equals("NY"))
          .mapToInt(t ‑> t.getAmount())
          .sum();

Operation filter() chỉ chọn các giao dịch với người bán từ New York. Operation mapToInt() chọn số tiền giao dịch cho các giao dịch mong muốn. Và terminal operation sum() sẽ cộng các số tiền này lại.

Mặc dù ví dụ này đẹp và dễ đọc, nhưng những người gièm pha có thể chỉ ra rằng phiên bản vòng lặp truyền thống(for-loop) của truy vấn này cũng đơn giản và cần ít dòng mã hơn để diễn đạt. Nhưng vấn đề không cần phải trở nên phức tạp hơn nhiều để những lợi ích của cách tiếp cận theo Stream trở nên rõ ràng. Streams khai thác nguyên tắc tính toán mạnh mẽ nhất: composition(thành phần). Bằng cách kết hợp các hoạt động phức tạp từ các khối xây dựng đơn giản (filtering, mapping, sorting, aggregation), các streams queries có nhiều khả năng vẫn dễ viết và đọc khi vấn đề trở nên phức tạp hơn so với các tính toán đặc biệt hơn trên cùng một nguồn dữ liệu.

Là một truy vấn phức tạp hơn từ cùng domain với Ví dụ 1, hãy xem xét "In tên của người bán trong các giao dịch với người mua trên 65 tuổi, được sắp xếp theo tên." Viết truy vấn này theo cách cũ (bắt buộc) có thể mang lại kết quả giống như Ví dụ 2.

Ví dụ 2

Set<Seller> sellers = new HashSet<>();
for (Txn t : txns) {
    if (t.getBuyer().getAge() >= 65)
        sellers.add(t.getSeller());
}
List<Seller> sorted = new ArrayList<>(sellers);
Collections.sort(sorted, new Comparator<Seller>() {
    public int compare(Seller a, Seller b) {
        return a.getName().compareTo(b.getName());
    }
});
for (Seller s : sorted)
    System.out.println(s.getName());

Mặc dù truy vấn này chỉ phức tạp hơn một chút so với truy vấn đầu tiên, nhưng rõ ràng là tổ chức và khả năng đọc mã kết quả theo cách tiếp cận cũ đã bắt đầu sụp đổ. Để đọc đoạn mã này, chúng ta cần phải ghi nhớ rất nhiều ngữ cảnh trước khi tìm ra mã thực sự làm gì. Ví dụ 3 cho thấy cách chúng ta có thể viết lại truy vấn này bằng Stream.

Ví dụ 3

txns.stream()
    .filter(t ‑> t.getBuyer().getAge() >= 65)
    .map(Txn::getSeller)
    .distinct()
    .sorted(comparing(Seller::getName))
    .map(Seller::getName)
    .forEach(System.out::println);

Mã trong ví dụ 3 dễ đọc hơn nhiều, bởi vì người dùng không bị phân tâm bởi các biến "rác" — như sellers và sortedp— và không phải theo dõi nhiều ngữ cảnh trong khi đọc mã; mã đọc gần như chính xác như bản chất thật sự của vấn đề. Mã dễ đọc hơn cũng ít bị lỗi hơn, bởi vì người bảo trì có nhiều khả năng có thể phân biệt chính xác chức năng của mã ngay từ cái nhìn đầu tiên.

Phương pháp thiết kế được thực hiện bởi các thư viện như Stream dẫn đến sự tách biệt thực tế giữa các mối quan tâm. Người sử dụng chịu trách nhiệm chỉ định "what(cái gì)" của phép tính, nhưng thư viện có quyền kiểm soát "how(như thế nào)". Sự tách biệt này có xu hướng song song với việc tách biệt về chuyên môn; người viết code thường hiểu rõ hơn về domain(bussines logic), trong khi người viết thư viện thường có chuyên môn hơn về các thuộc tính thuật toán của việc thực thi code. Yếu tố hỗ trợ chính để viết các thư viện cho phép loại mối quan tâm này là khả năng chuyển hành vi dễ dàng như truyền dữ liệu, từ đó cho phép các API nơi người gọi có thể mô tả cấu trúc của một phép tính phức tạp và sau đó không phải suy nghĩ xem bên trong thư viện thực thi như thế nào.

Anatomy of a stream pipeline(Giải phẫu stream pipeline)

Tất cả các tính toán stream đều có chung một cấu trúc: Chúng có một stream source, không hoặc nhiều intermediate operations và một terminal operation duy nhất. Các phần tử của luồng có thể là tham chiếu đối tượng (Stream<String>) hoặc chúng có thể là số nguyên nguyên thủy (IntStream), longs (LongStream) hoặc doubles (DoubleStream).

Bởi vì hầu hết dữ liệu mà các chương trình Java sử dụng đã được lưu trữ trong các collections, nên nhiều stream computations sử dụng các collections làm nguồn của chúng. Tất cả các triển khai Collection trong JDK đều đã được cải tiến để hoạt động như các nguồn stream hiệu quả. Nhưng các nguồn stream có thể khác cũng tồn tại — chẳng hạn như arrays, generator functions hoặc built-in factories như dãy số — và (như sẽ trình bày trong phần thứ ba của loạt bài này) có thể viết custom stream adapters để bất kỳ nguồn dữ liệu nào cũng có thể hoạt động như một nguồn stream. Bảng dưới cho thấy một số phương thức tạo stream trong JDK.

Intermediate operations - chẳng hạn như filter() (chọn các phần tử phù hợp với một tiêu chí), map() (chuyển đổi các phần tử theo chức năng), distinct() (loại bỏ các phần tử trùng lặp), limit() (cắt bớt luồng ở một kích thước cụ thể) và sorted() - chuyển đổi một luồng thành một luồng khác. Một số operations, chẳng hạn như mapToInt(), lấy một stream thuộc một loại và trả về một stream thuộc loại khác; ví dụ 1 bắt đầu dưới dạng Stream<Transaction> và sau đó chuyển sang IntStream. Bảng 2 cho thấy một số hoạt động của luồng trung gian.

Intermediate operations thì luôn luôn lazy: Việc gọi một Intermediate operations chỉ thiết lập giai đoạn tiếp theo trong stream pipeline chứ không bắt đầu bất kỳ công việc nào. Intermediate operations are further divided into được chia thành các hoạt động stateless và stateful. Stateless operations (như là filter() hay map()) có thể hoạt động độc lập trên từng phần tử, trong khi các stateful operations (như là sorted() hay distinct()) có thể kết hợp trạng thái từ các phần tử đã thấy trước đó ảnh hưởng đến quá trình xử lý các phần tử khác.

Quá trình xử lý tập dữ liệu bắt đầu khi một thao tác đầu cuối được thực thi, chẳng hạn như thao tác reduction (sum() hoặc max()), application (forEach()) hoặc search (findFirst()). Terminal operations tạo ra kết quả hoặc tác dụng phụ. Khi một terminal operation được thực thi, stream pipeline sẽ bị chấm dứt và nếu chúng ta muốn duyệt lại cùng một tập dữ liệu, chúng ta có thể thiết lập một stream pipeline.

Table 3. Terminal stream operations

Streams so với collections

Mặc dù bề ngoài các Streams có thể giống với các collections — chúng ta có thể nghĩ cả hai đều chứa dữ liệu — nhưng trên thực tế, chúng khác nhau đáng kể. Một collection là một cấu trúc dữ liệu; mối quan tâm chính của nó là tổ chức dữ liệu trong bộ nhớ và một bộ sưu tập vẫn tồn tại trong một khoảng thời gian. Một collection thường có thể được sử dụng làm nguồn hoặc đích cho một stream pipeline, nhưng trọng tâm của một stream's là computation(tính toán) chứ không phải data(dữ liệu). Dữ liệu đến từ nơi khác collection, array, generator function, hoặc I/O channel) và được xử lý thông qua một chuỗi các bước tính toán để tạo ra kết quả hoặc tác dụng phụ, tại thời điểm đó, stream kết thúc. Các Streams không cung cấp bộ lưu trữ cho các phần tử mà chúng xử lý và vòng đời của một Stream giống như một thời điểm — lệnh gọi của terminal operation. Không giống như collections, các stream cũng có thể là vô hạn; tương ứng, một số operations (limit(), findFirst()) là short-circuiting(ngắt mạch) và có thể hoạt động trên các infinite streams(luồng vô hạn) với tính toán hữu hạn.

Collections và streams cũng khác nhau trong cách mà các operations của chúng được thực thi. Operations trên các collections là eager(háo hức) và mutative(đột biến); khi phương thức remove() được gọi trong List, sau khi cuộc gọi trả về, chúng ta biết rằng trạng thái List đã được sửa đổi để phản ánh việc loại bỏ phần tử đã chỉ định. Đối với các streams, chỉ terminal operation là eager(háo hức); còn lại là lazy. Các Stream operations biểu thị một phép biến đổi chức năng trên đầu vào của chúng (cũng là một Stream), chứ không phải là một thao tác biến đổi trên một tập dữ liệu (filter một stream tạo ra một stream mới có các phần tử là tập hợp con của stream đầu vào nhưng không loại bỏ bất kỳ phần tử nào khỏi nguồn ban đầu).

Việc thể hiện một stream pipeline dưới dạng một chuỗi các phép biến đổi chức năng cho phép một số chiến lược thực thi hữu ích, chẳng hạn như laziness, short-circuiting(ngắt mạch) và các operation fusion(hợp nhất). Short-circuiting cho phép một pipeline kết thúc thành công mà không cần kiểm tra tất cả dữ liệu; các truy vấn như "tìm giao dịch đầu tiên trên 1.000 đô la" không cần kiểm tra thêm bất kỳ giao dịch nào sau khi tìm thấy kết quả khớp. Operation fusion có nghĩa là nhiều thao tác có thể được thực hiện trong một lần truyền dữ liệu; trong ví dụ 1, ba thao tác được kết hợp thành một lần truyền dữ liệu — thay vì trước tiên chọn tất cả các giao dịch phù hợp, sau đó chọn tất cả số tiền tương ứng, rồi cộng chúng lại.

Phiên bản sử dụng code kiểu truyền thống của các truy vấn như trong Ví dụ 1 và Ví dụ 3 thường sử dụng các tập hợp cụ thể hóa cho kết quả của các phép tính trung gian, chẳng hạn như kết quả của filter hoặc mapping. Những kết quả này không chỉ có thể làm lộn xộn mã mà còn làm lộn xộn quá trình thực thi. Việc cụ thể hóa các collections trung gian chỉ phục vụ cho việc implementation chứ không phải kết quả và nó sử dụng các chu kỳ tính toán để tổ chức các kết quả trung gian thành các cấu trúc dữ liệu sẽ chỉ bị loại bỏ.

Ngược lại, các Stream pipeline hợp nhất các hoạt động của chúng thành ít lần truyền dữ liệu nhất có thể, thường là một lần truyền. (Stateful intermediate operations, chẳng hạn như sorting, có thể đưa ra các điểm rào cản bắt buộc phải thực thi nhiều lần.) Mỗi giai đoạn của stream pipeline tạo ra các phần tử của nó một cách lazy, chỉ tính toán các phần tử khi cần và đưa chúng trực tiếp đến giai đoạn tiếp theo. chúng ta không cần một collection để giữ kết quả trung gian của quá trình filtering hoặc mapping, vì vậy chúng ta tiết kiệm được công sức điền vào (và garbage collecting-thu gom rác) các collections trung gian. Ngoài ra, việc tuân theo chiến lược thực thi "depth first" thay vì "breadth first" (theo dõi đường dẫn của một phần tử dữ liệu duy nhất trong toàn bộ đường dẫn) khiến dữ liệu được vận hành thường xuyên bị "nóng" hơn trong bộ đệm, vì vậy chúng ta có thể chi tiêu nhiều thời gian tính toán hơn và ít thời gian chờ dữ liệu hơn.

Ngoài việc sử dụng các streams để tính toán, chúng ta có thể cân nhắc sử dụng các streams để trả về từ các phương thức API, nơi mà trước đây chúng ta có thể đã trả về một mảng hoặc tập hợp. Trả về một luồng thường hiệu quả hơn vì chúng ta không phải sao chép tất cả dữ liệu vào một mảng hoặc bộ sưu tập mới. Trả lại một stream cũng thường linh hoạt hơn; các form của collection mà thư viện chọn trả về có thể không phải là thứ mà người gọi cần và thật dễ dàng để chuyển đổi một stream thành bất kỳ loại collection nào. (Tình huống chính trong đó việc trả về một stream là không phù hợp và việc quay lại trả về một collection cụ thể hóa sẽ tốt hơn, đó là khi người gọi cần xem snapshot nhất quán về trạng thái tại một thời điểm.)

Parallelism(Song song)

Hệ quả hữu ích của việc cấu trúc tính toán dưới dạng các phép biến đổi hàm là chúng ta có thể dễ dàng chuyển đổi giữa thực thi tuần tự và song song với những thay đổi tối thiểu đối với mã. Biểu thức tuần tự của tính toán luồng và biểu thức song song của cùng một phép tính gần như giống hệt nhau. Ví dụ 4 cho thấy cách thực hiện song song truy vấn từ Ví dụ 1.

Ví dụ 4

int totalSalesFromNY
    = txns.parallelStream()
          .filter(t ‑> t.getSeller().getAddr().getState().equals("NY"))
          .mapToInt(t ‑> t.getAmount())
          .sum();

Yêu cầu của dòng đầu tiên đối với mộtparallel stream thay vì một sequential stream là điểm khác biệt duy nhất so với Ví dụ 1. Trước đây, chạy song song đòi hỏi phải viết lại hoàn toàn mã, điều này không chỉ tốn kém mà còn thường dễ bị lỗi, vì mã song song thu được trông không giống phiên bản tuần tự.

Tất cả các stream operations có thể được thực hiện tuần tự hoặc song song, nhưng hãy nhớ rằng tính song song không phải là hiệu suất ma thuật. Thực thi song song có thể nhanh hơn, cùng tốc độ hoặc chậm hơn so với thực thi tuần tự. Tốt nhất là bắt đầu với các sequential streams và áp dụng tính parallelism khi chúng ta biết rằng khi nào chúng ta sẽ nhận được — và hưởng lợi từ — một sự tăng tốc của parallelism. Phần sau của loạt bài này sẽ quay lại phân tích một stream pipeline để có hiệu suất song song.

The fine print(Những ràng buộc phải tuân theo)

Vì thư viện Streams đang điều phối quá trình tính toán, nhưng việc thực hiện tính toán bao gồm các cuộc gọi lại tới lambda do chúng ta truyền vào, nên những biểu thức lambda đó có thể thực hiện phải tuân theo một số ràng buộc nhất định. Việc vi phạm các ràng buộc này có thể khiến stream pipeline bị lỗi hoặc tính toán kết quả không chính xác. Ngoài ra, đối với lambda có tác dụng phụ, thời gian (hoặc sự tồn tại) của những tác dụng phụ này có thể gây ngạc nhiên trong một số trường hợp.

Hầu hết cácstream operations đều yêu cầu lambda được truyền cho chúng là non-interfering(không can thiệp) và stateless. Non-interfering(Không can thiệp) có nghĩa là họ sẽ không sửa đổi nguồn của stream; stateless có nghĩa là chúng sẽ không truy cập (đọc hoặc ghi) bất kỳ trạng thái nào có thể thay đổi trong suốt thời gian hoạt động của stream. Đối với reduction operations (ví dụ: tính toán dữ liệu như sum, min hoặc tối đa), lambda được truyền vào cho các hoạt động này phải là associative (hoặc tuân theo các yêu cầu tương tự).

Các yêu cầu này một phần xuất phát từ thực tế là thư viện Strean có thể, nếu pipeline thực thi song song, truy cập nguồn dữ liệu hoặc gọi các lambda này đồng thời từ nhiều luồng. Các hạn chế là cần thiết để đảm bảo rằng tính toán vẫn chính xác. (Những hạn chế này cũng có xu hướng dẫn đến mã đơn giản hơn, dễ hiểu hơn, bất kể tính song song.) chúng ta có thể muốn thuyết phục bản thân rằng chúng ta có thể bỏ qua những hạn chế này bởi vì chúng ta không nghĩ rằng một quy trình cụ thể sẽ chạy song song, nhưng tốt nhất là chúng ta nên chống lại sự cám dỗ này, nếu không chúng ta sẽ chôn bom hẹn giờ trong mã của mình. Hãy nỗ lực thể hiện các stream pipelines của chúng ta sao cho chúng sẽ chính xác bất kể chiến lược thực thi là gì.

Nguồn gốc của tất cả các rủi ro tương tranh này là shared mutable state(trạng thái có thể thay đổi được chia sẻ). Một nguồn có trạng thái có thể thay đổi được chia sẻ là stream source. Nếu nguồn là một collection truyền thống như ArrayList, thì thư viện Streams giả định rằng nó không bị sửa đổi trong quá trình hoạt động của luồng. (Các Collections được thiết kế rõ ràng để truy cập đồng thời, chẳng hạn như ConcurrentHashMap, được miễn trừ khỏi giả định này.) Yêu cầu không can thiệp không chỉ loại trừ nguồn bị biến đổi bởi các threads khác trong quá trình stream hoạt động, mà bản thân các lambda được truyền cho stream hoạt động cũng phải hạn chế làm biến đổi nguồn. Ngoài việc không sửa đổi stream source, lambdas được chuyển đến stream operations phải là stateless. Ví dụ 5: mã cố gắng loại bỏ bất kỳ phần tử nào gấp đôi phần tử trước đó, vi phạm quy tắc này.

Ví dụ 5

HashSet<Integer> twiceSeen = new HashSet<>();
int[] result
    = elements.stream()
              .filter(e ‑> {
                  twiceSeen.add(e * 2);
                  return twiceSeen.contains(e);
              })
              .toArray();

Nếu được thực thi song song, quy trình này sẽ tạo ra kết quả không chính xác vì hai lý do. Đầu tiên, quyền truy cập vào twiceSeen được thực hiện từ nhiều luồng mà không có bất kỳ sự phối hợp nào và do đó luồng không an toàn. Thứ hai, vì dữ liệu được phân vùng nên không đảm bảo rằng khi một phần tử nhất định được xử lý, tất cả các phần tử trước phần tử đó đã được xử lý.

Tốt nhất là các lambda được truyền vào stream operations hoàn toàn không có tác dụng phụ— nghĩa là chúng không làm thay đổi bất kỳ trạng thái trên bộ nhớ heap nào hoặc thực hiện bất kỳ I/O đọc ghi nào trong quá trình thực thi. Nếu nó có tác dụng phụ, thì nó phải có trách nhiệm cung cấp bất kỳ sự phối hợp cần thiết nào để đảm bảo rằng các tác dụng phụ đó là thread safe(luồng an toàn).

Hơn nữa, thậm chí không đảm bảo rằng tất cả các tác dụng phụ sẽ được thực thi. Ví dụ 6, thư viện được tự do tránh hoàn toàn việc thực thi lambda được truyền tới map(). Vì nguồn có kích thước đã biết, thao tác map() được biết là bảo toàn kích thước và ánh xạ không ảnh hưởng đến kết quả tính toán, thư viện có thể tối ưu hóa phép tính bằng cách hoàn toàn không thực hiện mapping! (Việc tối ưu hóa này có thể chuyển phép tính từ O(n) sang O(1), ngoài ra còn loại bỏ công việc liên quan đến việc gọi hàm mapping).

Ví dụ 6

int count = 
    anArrayList.stream()
               .map(e ‑> { System.out.println("Saw " + e); e })
               .count();

Trường hợp duy nhất mà chúng ta sẽ nhận thấy hiệu quả của việc tối ưu hóa này (ngoài việc tính toán nhanh hơn đáng kể) là nếu lambda được chuyển đến map() có tác dụng phụ - trong trường hợp đó chúng ta có thể ngạc nhiên nếu những tác dụng phụ đó không xảy ra. Việc có thể thực hiện các tối ưu hóa này dựa trên giả định rằng các hoạt động của luồng là các phép biến đổi chức năng. Hầu hết thời gian, chúng ta thích thư viện làm cho mã của chúng ta chạy nhanh hơn mà không cần nỗ lực từ phía chúng ta. Cái giá phải trả để có thể thực hiện các tối ưu hóa như thế này là chúng ta phải chấp nhận một số hạn chế về những gì lambdas mà chúng ta truyền vào các stream operations có thể thực hiện và về một số sự phụ thuộc của chúng ta vào các tác dụng phụ. (Nhìn chung, đây là một giao dịch khá tốt.)

Kết luận cho Phần 1

Thư viện java.util.stream cung cấp một phương tiện đơn giản và linh hoạt để thể hiện các truy vấn kiểu chức năng có thể song song trên nhiều nguồn dữ liệu khác nhau, bao gồm các collections, arrays, generator functions hoặc cấu trúc dữ liệu tùy chỉnh khác. Khi chúng ta bắt đầu sử dụng nó, chúng ta sẽ bị cuốn hút!