Giới thiệu

Trong nhiều tình huống, chúng ta cần lấy một lượng lớn dữ liệu nhưng thường gặp phải sự chậm trễ trước khi nhận được phản hồi đầu tiên. May mắn thay, có một giải pháp hiệu quả cho vấn đề này thông qua việc sử dụng Streaming Response trong Spring.

TechStack

Dự án này được xây dựng với:

• Java 24.
• Spring Boot 3.5.5 với WebFlux.
• Postgres (hoặc bất kỳ cơ sở dữ liệu nào bạn muốn).

Trường hợp sử dụng

Chúng ta cần có 1 triệu sản phẩm (đối tượng JSON) trong một endpoint GET.

Dưới đây là một script SQL đơn giản để tạo và điền dữ liệu cho bảng sản phẩm:

-- Tạo bảng
CREATE TABLE IF NOT EXISTS products (
    id SERIAL PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    price NUMERIC(10,2) NOT NULL
);

-- Chèn 1 triệu sản phẩm
INSERT INTO products (name, price)
SELECT
    'Sản phẩm ' || i,
    ROUND((RANDOM() * 1000)::numeric, 2)
FROM generate_series(1, 1000000) AS s(i);

Thực thể và JPA Repository

Dưới đây là định nghĩa cho thực thể sản phẩm và repository:

@Entity
@Table(name = "products")
public class Product {
    @Id
    private Long id;
    private String name;
    private Double price;

    public Product() {
    }

    public Product(Long id, String name, Double price) {
        this.id = id;
        this.name = name;
        this.price = price;
    }

    public Long getId() {
        return id;
    }

    public String getName() {
        return name;
    }

    public Double getPrice() {
        return price;
    }

    @Override
    public String toString() {
        return "Product[" +
                "id=" + id + ", " +
                "name=" + name + ", " +
                "price=" + price + ']';
    }
}

import org.springframework.data.jpa.repository.JpaRepository;

public interface ProductRepository extends JpaRepository<Product, Long> {
}

Bạn có thể sử dụng bất kỳ cơ sở dữ liệu nào ở đây, tôi chỉ quen với Postgres do công việc hàng ngày của mình.

Endpoint truyền thống

Chúng ta sẽ có một endpoint GET đơn giản để tìm tất cả sản phẩm:

@RestController
@RequestMapping("/products")
public class ProductController {

    private final ProductRepository productRepository;

    public ProductController(ProductRepository productRepository) {
        this.productRepository = productRepository;
    }

    @GetMapping
    public List<Product> getAll(){
       return productRepository.findAll();
    }
}

Khi sử dụng endpoint này, khách hàng sẽ phải chờ đợi trước khi nhận được sản phẩm đầu tiên, hoặc tệ hơn, kích thước phản hồi quá lớn để xử lý một cách hợp lý.

Streaming Response giúp khắc phục vấn đề

Bằng cách sử dụng StreamingResponseBody, chúng ta có thể stream các sản phẩm một cách tuần tự và tiết kiệm thời gian:

@GetMapping("/stream")
public StreamingResponseBody getAllStreamed(HttpServletResponse response){
    response.setContentType("text/event-stream");
    return outputStream -> {
        productRepository.findAll()
            .forEach(product -> {
                try {
                    String json = new ObjectMapper().writeValueAsString(product) + "\n";
                    outputStream.write(json.getBytes());
                    outputStream.flush();
                } catch (JsonProcessingException e) {
                    log.error("Lỗi khi chuyển đổi sản phẩm thành JSON", e);
                } catch (IOException e) {
                    log.error("Lỗi khi ghi đối tượng vào stream", e);
                }
            });
    };
}

Tối ưu hóa thêm

Bạn có thể đã nhận ra rằng chúng ta lặp qua các sản phẩm và ghi từng sản phẩm vào output stream. Điều này có thể được cải thiện hơn nữa bằng cách streaming dữ liệu từ nguồn đến client mà không cần qua bước trung gian. Để làm điều này, chúng ta có thể sử dụng các giải pháp như Spring WebFlux và JPA Streams.

Thực hành tốt nhất

• Sử dụng StreamingResponseBody: Đây là cách hiệu quả nhất để xử lý các lượng lớn dữ liệu mà không làm tắc nghẽn bộ nhớ.
• Xem xét sử dụng Spring WebFlux: Để xử lý các yêu cầu không đồng bộ và tối ưu hóa hiệu suất ứng dụng.

Cạm bẫy thường gặp

• Xử lý lỗi: Luôn xử lý các ngoại lệ khi làm việc với stream để tránh việc ứng dụng bị treo.
• Kích thước phản hồi: Đảm bảo rằng kích thước phản hồi không quá lớn để tránh gây ra độ trễ.

Mẹo hiệu suất

• Sử dụng JPA Stream: Khi lấy dữ liệu lớn, hãy sử dụng Stream<T> để tránh tải toàn bộ dữ liệu vào bộ nhớ.
• Kiểm tra tốc độ: Đo lường thời gian phản hồi và điều chỉnh số lượng sản phẩm trả về trong mỗi lần stream.

Câu hỏi thường gặp (FAQ)

1. StreamingResponseBody là gì?

StreamingResponseBody là một tính năng trong Spring cho phép trả về dữ liệu từ server đến client theo dạng stream.

2. Làm thế nào để xử lý lỗi trong StreamingResponseBody?

Bạn nên luôn sử dụng các khối try-catch để xử lý các ngoại lệ có thể xảy ra trong quá trình ghi dữ liệu vào stream.

Tài nguyên tham khảo

Streaming với JPA Stream<T>

• Cách tốt nhất để sử dụng phương thức Stream trong Spring Data JPA

Server-Sent Events (SSE) / Reactive

• Tài liệu Spring Boot WebFlux
• Hướng dẫn Project Reactor Flux

Hãy đọc bài viết này và nhiều bài khác trên trang web của tôi: https://www.saadelattar.me/article/spring-response-streams