Giới thiệu về Async Generator

Async Generator là một thư viện mạnh mẽ trong hệ sinh thái Java, được thiết kế để giải quyết bài toán xử lý luồng dữ liệu bất đồng bộ (Asynchronous Streams). Nó kết hợp khả năng tính toán lười biếng (lazy evaluation) với khả năng lặp có thể ngắt quãng, giúp lập trình viên quản lý các tác vụ non-blocking một cách hiệu quả hơn.

Thư viện này đặc biệt hữu ích trong các kịch bản như:

• Xử lý kết quả từ các API gọi hàng loạt (Batch API).
• Làm việc với các luồng tin nhắn (Message Streams) hoặc sự kiện (Event Streams).
• Truy vấn dữ liệu phân trang từ cơ sở dữ liệu.
• Thực hiện các thao tác biến đổi dữ liệu trên luồng bất đồng bộ mà không cần tải toàn bộ dữ liệu vào bộ nhớ.

Cài đặt thư viện

Để bắt đầu sử dụng, bạn cần thêm dependency sau vào tệp pom.xml của dự án Maven:

<dependency>
    <groupId>org.bsc.async</groupId>
    <artifactId>async-generator</artifactId>
    <version>4.2.0</version>
</dependency>

Các thành phần cốt lõi

Thành phần	Mô tả chức năng
AsyncGenerator<T>	Giao diện chính để định nghĩa và thao tác với luồng bất đồng bộ.
AsyncGeneratorQueue<T>	Một dạng generator dựa trên hàng đợi, cho phép đẩy dữ liệu vào luồng một cách linh hoạt.
AsyncGenerator.Data<T>	Lớp bao bọc chứa giá trị trả về, trạng thái hoàn tất hoặc ngoại lệ.
WithEmbed<T>	Tiện ích hỗ trợ việc gộp hoặc nối chuỗi nhiều generator lại với nhau.

Khởi tạo Async Generator

Có nhiều cách để tạo ra một luồng AsyncGenerator tùy thuộc vào nguồn dữ liệu đầu vào.

1. Tạo từ một Collection

Phương thức này chuyển đổi một danh sách có sẵn thành các tác vụ bất đồng bộ.

List<String> skuList = List.of("PROD-01", "PROD-02", "PROD-03");

AsyncGenerator<String> productGen = AsyncGenerator.map(
    skuList,
    sku -> CompletableFuture.supplyAsync(() -> "Detail of " + sku)
);

2. Tạo từ danh sách CompletableFuture

CompletableFuture<Integer> task1 = CompletableFuture.completedFuture(100);
CompletableFuture<Integer> task2 = CompletableFuture.completedFuture(200);

AsyncGenerator<Integer> numGen = AsyncGenerator.from(task1, task2);

3. Sử dụng Queue để đẩy dữ liệu động

Đây là cách tiếp cận lý tưởng cho mô hình Producer-Consumer, nơi dữ liệu được sản sinh liên tục.

AsyncGeneratorQueue<Double> priceQueue = AsyncGeneratorQueue.create();

// Đẩy dữ liệu vào queue
priceQueue.offer(CompletableFuture.completedFuture(19.99));
priceQueue.offer(CompletableFuture.completedFuture(25.50));

// Thông báo luồng kết thúc (quan trọng để tránh treo luồng lặp)
priceQueue.complete();

Thao tác duyệt dữ liệu

Duyệt bất đồng bộ (Non-blocking)

Phương thức forEachAsync giúp xử lý từng phần tử ngay khi nó sẵn sàng mà không chặn luồng chính.

productGen.forEachAsync(data -> {
    System.out.println("Processing: " + data);
}).join(); // Chờ toàn bộ luồng kết thúc

Duyệt đồng bộ (Blocking)

Bạn có thể sử dụng vòng lặp for-each truyền thống, luồng sẽ đợi từng phần tử hoàn tất trước khi sang bước tiếp theo.

for (String item : productGen) {
    System.out.println("Received: " + item);
}

Các toán tử biến đổi luồng

Tương tự như Java Stream API, Async Generator cung cấp các phương thức để thao tác trên dữ liệu.

• map: Biến đổi kiểu dữ liệu của từng phần tử.

  AsyncGenerator<Integer> lengthGen = productGen.map(String::length);

• filter: Lọc các phần tử theo điều kiện.

  AsyncGenerator<String> filteredGen = productGen.filter(s -> s.endsWith("01"));

• take: Giới hạn số lượng phần tử lấy ra.

  AsyncGenerator<String> limitedGen = productGen.take(5);

• flatMap: Trải phẳng các generator lồng nhau.

  AsyncGenerator<String> flattened = productGen.flatMap(s -> 
      AsyncGenerator.from(CompletableFuture.completedFuture(s + " - Part A"))
  );

Xử lý lỗi và Ngắt luồng

Thư viện cho phép bắt ngoại lệ của từng tác vụ riêng lẻ mà không làm sập toàn bộ quy trình xử lý luồng.

CompletableFuture<String> errorTask = CompletableFuture.failedFuture(new RuntimeException("API Timeout"));
AsyncGenerator<String> genWithErr = AsyncGenerator.from(errorTask);

genWithErr.forEachAsync(
    System.out::println, 
    err -> System.err.println("Handled error: " + err.getMessage())
).join();

Để hủy bỏ việc thực thi luồng giữa chừng, bạn có thể sử dụng phương thức cancel trên đối tượng Future trả về từ forEachAsync.

CompletableFuture<Void> execution = productGen.forEachAsync(System.out::println);
execution.cancel(true); // Ngắt tiến trình

Ví dụ thực tế: Gọi API lấy thông tin kho hàng

Dưới đây là một ví dụ mô phỏng việc lấy dữ liệu từ nhiều dịch vụ khác nhau một cách bất đồng bộ.

import org.bsc.async.AsyncGenerator;
import java.util.Arrays;
import java.util.concurrent.CompletableFuture;

public class InventoryService {

    private static CompletableFuture<Integer> checkStock(String productId) {
        return CompletableFuture.supplyAsync(() -> {
            // Giả lập độ trễ mạng
            try { Thread.sleep(200); } catch (InterruptedException e) {}
            return (int) (Math.random() * 100);
        });
    }

    public static void main(String[] args) {
        var ids = Arrays.asList("IPHONE-15", "MACBOOK-M3", "AIRPODS-PRO");

        AsyncGenerator<Integer> stockGen = AsyncGenerator.map(ids, InventoryService::checkStock);

        System.out.println("Bắt đầu kiểm tra kho...");

        stockGen.forEachAsync(stock -> {
            System.out.println("Số lượng tồn kho: " + stock);
        }).join();

        System.out.println("Hoàn thành báo cáo.");
    }
}

Quy tắc cần lưu ý

• Tính chất lười biếng: Các tác vụ bất đồng bộ sẽ không được thực thi cho đến khi bạn bắt đầu duyệt qua generator.
• Kết thúc hàng đợi: Đối với AsyncGeneratorQueue, luôn luôn phải gọi complete() để giải phóng tài nguyên.
• Lựa chọn vòng lặp: Luôn ưu tiên forEachAsync để tận dụng tối đa hiệu năng của lập trình non-blocking.
• Ghép nối: Sử dụng withEmbed khi cần kết hợp dữ liệu từ nhiều nguồn generator khác nhau thành một luồng duy nhất.