Spring Integration là một framework mạnh mẽ triển khai các mô hình tích hợp hệ thống doanh nghiệp (Enterprise Integration Patterns - EIP). Dựa trên nền tảng Spring Messaging, nó mở rộng mô hình lập trình của Spring để hỗ trợ việc truyền dẫn dữ liệu giữa các thành phần hoặc hệ thống khác nhau thông qua cơ chế hướng sự kiện (event-driven).

Các đặc điểm chính của Spring Integration

• Kiến trúc hướng thông điệp: Sử dụng Message làm thực thể trung gian để vận chuyển dữ liệu, giúp các thành phần giao tiếp mà không cần biết trực tiếp về nhau.
• Sử dụng Channel và Filter: Channel đóng vai trò là đường ống dẫn, trong khi Filter xử lý logic lọc hoặc điều hướng thông điệp.
• Đa dạng phương thức giao tiếp: Hỗ trợ từ Point-to-Point (điểm tới điểm) đến Publish-Subscribe (xuất bản - đăng ký), bao gồm cả xử lý đồng bộ và bất đồng bộ.
• Khả năng mở rộng cao: Cung cấp các adapter có sẵn để kết nối với File, JMS, HTTP, Email, MQTT và nhiều giao thức khác.
• Tích hợp chặt chẽ với Spring Ecosystem: Tận dụng tối đa Dependency Injection, quản lý Transaction và Spring Boot auto-configuration.

Cấu trúc cơ bản của hệ thống

1. Message (Thông điệp)

Trong Spring Integration, Message là một container bao bọc dữ liệu Java. Nó bao gồm hai phần chính:

• Payload: Dữ liệu thực tế cần truyền tải (có thể là bất kỳ đối tượng Java nào).
• Headers: Metadata đi kèm như ID, Timestamp, Correlation ID, hoặc các thông tin tùy chỉnh như định dạng file, thông tin xác thực.

2. Message Channel (Kênh dẫn)

Channel đóng vai trò là "đường ống" nối giữa bên gửi (Producer) và bên nhận (Consumer), giúp tách biệt hoàn toàn sự phụ thuộc giữa chúng.

Các loại Channel phổ biến:

• DirectChannel: Đây là loại mặc định. Nó gửi thông điệp trực tiếp đến Subscriber trong cùng một luồng (thread) của bên gửi. Nó không có hàng đợi (buffer).
• QueueChannel: Sử dụng hàng đợi để lưu trữ thông điệp. Bên nhận sẽ lấy dữ liệu thông qua cơ chế Polling (thăm dò).
• PublishSubscribeChannel: Gửi một thông điệp đến tất cả các bên đang đăng ký kênh đó (Broadcasting).
• ExecutorChannel: Tương tự DirectChannel nhưng việc gửi thông điệp được thực hiện thông qua một TaskExecutor (Thread Pool), hỗ trợ xử lý bất đồng bộ.

// Ví dụ khởi tạo một QueueChannel với sức chứa giới hạn
@Bean
public MessageChannel dataProcessingChannel() {
    return new QueueChannel(500);
}

// Đăng ký xử lý logic cho một SubscribableChannel
@Bean
public MessageChannel notificationChannel() {
    DirectChannel channel = new DirectChannel();
    channel.subscribe(message -> {
        System.out.println("Nhận payload: " + message.getPayload());
    });
    return channel;
}

3. Channel Interceptor (Trình chặn kênh)

Interceptor cho phép chúng ta can thiệp vào quá trình gửi hoặc nhận thông điệp trên Channel mà không làm thay đổi logic nghiệp vụ chính.

public class CustomLoggingInterceptor implements ChannelInterceptor {
    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        // Thực hiện logic kiểm tra hoặc log trước khi gửi
        return message; 
    }
}

Các điểm cuối thông điệp (Message Endpoints)

Message Transformer

Transformer có nhiệm vụ thay đổi nội dung hoặc cấu trúc của thông điệp, ví dụ chuyển đổi từ JSON String sang một Object cụ thể.

@Transformer(inputChannel = "jsonInputChannel", outputChannel = "objectOutputChannel")
public UserProfile parseUserJson(String jsonRaw) {
    return jsonMapper.readValue(jsonRaw, UserProfile.class);
}

Message Filter

Filter quyết định xem một thông điệp có được phép đi tiếp đến kênh đầu ra hay không dựa trên một điều kiện logic (Boolean).

@Filter(inputChannel = "orderChannel", outputChannel = "priorityOrderChannel")
public boolean isHighValueOrder(Order order) {
    return order.getAmount() > 1000;
}

Message Router

Router phân tích thông điệp để quyết định kênh tiếp theo mà thông điệp đó sẽ được gửi tới.

@Router(inputChannel = "routingChannel")
public String routeByCustomerType(Customer customer) {
    return customer.isVip() ? "vipServiceChannel" : "standardServiceChannel";
}

Splitter và Aggregator

• Splitter: Chia một thông điệp lớn (ví dụ: một danh sách đơn hàng) thành nhiều thông điệp nhỏ để xử lý song song.
• Aggregator: Gom nhiều thông điệp riêng lẻ lại thành một thông điệp duy nhất (ngược lại với Splitter).

@Splitter(inputChannel = "batchInbound", outputChannel = "singleItemChannel")
public List<OrderItem> splitOrder(Order order) {
    return order.getItems();
}

Service Activator

Đây là thành phần kết nối hệ thống tin nhắn với một bean dịch vụ cụ thể. Khi có thông điệp đến, Service Activator sẽ gọi phương thức của bean đó.

@ServiceActivator(inputChannel = "emailRequestChannel")
public void sendEmailNotification(EmailContent content) {
    emailService.send(content);
}

Channel Adapter

Adapter là cầu nối giữa Spring Integration và thế giới bên ngoài:

• Inbound Adapter: Đưa dữ liệu từ bên ngoài (File, DB, API) vào hệ thống tin nhắn.
• Outbound Adapter: Đẩy dữ liệu từ hệ thống tin nhắn ra bên ngoài.