# Event-driven trong Microservices ## 1. Giới thiệu ### 1.1. Temporal Coupling Cần tránh `Temporal Coupling` (phụ thuộc thời gian) giữa các service bất cứ khi nào có thể: - `Temporal Coupling` xảy ra khi một `service gọi` mong muốn phản hồi ngay lập tức từ `service được gọi` trước khi tiếp tục xử lý công việc của nó. Nếu `service được gọi` gặp bất kỳ sự chậm trễ nào trong việc phản hồi, điều đó sẽ làm ảnh hưởng tiêu cực tới thời gian phản hồi chung của `service gọi`. - Kịch bản này thường xuyên xảy ra trong quá trình trao đổi dữ liệu giữa các service trong một hệ thống Microservices. ### 1.2. Asynchronous Communication `Syncronous Communication` (giao tiếp đồng bộ) giữa các service không phải lúc nào cũng là lựa chọn tốt. Trong nhiều tình huống thực tế, `Asynchronous Communication` (giao tiếp bất đồng bộ) là lựa chọn tốt hơn. `Asynchronous Communication` có thể đáp ứng các yêu cầu một cách hiệu quả. ### 1.3. Building event-driven Microservices Một sự kiện, cũng như một sự cố, biểu thị một sự kiện quan trọng xảy ra trong hệ thống. Sự kiện có thể là bất kỳ hành động nào mà một service thực hiện, hoặc một sự kiện ngoại lệ xảy ra trong hệ thống, như việc chuyển đổi các trạng thái của một service. Nhiều nguồn có thể tạo ra sự kiện. Khi sự kiện xảy ra, nó có thể cảnh báo cho các bên liên quan. `Event-driven Microservices` có thể được xây dựng nhờ sử dụng: - `Event-driven Architecture`: là một kiến trúc phần mềm mà các thành phần của nó giao tiếp với nhau bằng cách gửi và nhận các sự kiện. - `Producing and consuming events using Asynchronous Communication`: các service có thể tạo ra sự kiện và tiêu thụ sự kiện sử dụng giao tiếp bất đồng bộ. - `Event brokers`: là một phần mềm giữa các service, giúp chuyển tiếp sự kiện từ service này sang service khác. - `Spring Cloud Function`: là một dự án Spring Cloud giúp xây dựng các service dựa trên các hàm Spring. - `Spring Cloud Stream`: là một dự án Spring Cloud giúp xây dựng các service dựa trên các luồng dữ liệu. #### 1.3.1. Event-driven Architecture `Event-driven Architecture` là một kiến trúc phần mềm mà các thành phần của nó giao tiếp với nhau bằng cách gửi và nhận các sự kiện. Có `2` loại mô hình chính trong `Event-driven Architecture`: - `Publisher/Subcriber (Pub/Sub) Model`: Mô hình này sẽ xoay quanh việc `subcription`. `Producer` sẽ tạo ra các sự kiện, nó được phân phối tới tất cả các `subcriber` để tiêu thụ. Khi 1 `event` đã được nhận, nó không thể được phân phối lại nữa, nghĩa là những `subcriber` tham gia sau khi `event` đã được phân phối sẽ không nhận được các `event` đã được phân phối trước đó (`pass event`). Mô hình này được sử dụng trong **`RABBITMQ`**. - `Event Streaming Model`: Trong mô hình này, các `event` sẽ được ghi vào `log` một cách tuần tự. `Producer` xuất bản các `event` khi chúng xảy ra và các `event` này sẽ được lưu trữ theo thứ tự 1 cách hợp lý. Thay vì việc phải `subcribe` vào các `event`, `consumer` có thể đọc từ bất kỳ phần nào của `event stream`. Một lợi thế của mô hình này là `event` có thể được phát, đọc và xử lý nhiều lần, cho phép `client` tham gia bất kỳ lúc nào cũng có thể đọc được `event` đã được phát trước đó. Mô hình này được sử dụng trong **`KAFKA`**. #### 1.3.2. RabbitMQ `RabbitMQ` là một `message broker` mã nguồn mở được xây dựng trên `Erlang`. Nó được công nhận rộng rãi nhờ sử dụng `AMQP` (Advanced Message Queuing Protocol) làm giao thức truyền thông và khả năng cung cấp các `asynchronous messaging` linh hoạt, triển khai phân tán và giám sát toàn diện. Khi sử dụng các solution dựa trên `AMQP`, `RabbitMQ` sẽ phân loại những bên tham gia thành các đối tượng sau: - `Producer`: là bên tạo ra `message` và gửi nó tới `RabbitMQ` (publisher). - `Consumer`: là bên nhận `message` từ `RabbitMQ` (subscriber). - `Message Broker`: là `RabbitMQ` server chịu trách nhiệm nhận và phân phối `message` giữa `producer` và `consumer`. Mô hình `AMQP` hoạt động dựa trên nguyên tắc `exchange` và `queue`: 1. `Producer` sẽ gửi `message` tới `exchange`. 2. `exchange` sẽ định tuyến `message` tới `queue` theo các quy tắc đã được cấu hình, và bản sao của `message` sẽ được lưu trữ trong `queue`. 3. `consumer` sẽ nhận `message` từ `queue`. Mỗi `message` chỉ được `consumer` nhận một lần và mỗi một `consumer` chỉ nhận `message` từ một `queue`. #### 1.3.3. Spring Cloud Function `Spring Cloud Function` là một dự án Spring Cloud giúp xây dựng các service dựa trên các hàm Spring, tuân thủ theo các giao diện tiêu chuẩn như `Supplier`, `Function`, `Consumer`. - `Supplier`: là một hàm `không nhận tham số đầu vào` và `trả về một giá trị`. - `Function`: là một hàm `nhận một tham số đầu vào` và `trả về một giá trị`. Thường được gọi là một bộ xử lý. - `Consumer`: là một hàm `nhận một tham số đầu vào` và `không trả về giá trị`. Thường được gọi là thuê bao hoặc bộ tiêu thụ. Các tính năng chính của `Spring Cloud Function`: - Lựa chọn về kiểu lập trình như `reactive`, `imperative`, `hybrid`. - `POJO Functions`: nếu có gì đó phù hợp với ngữ nghĩa `@FunctionalInterface` thì nó có thể được sử dụng như một `Function`. - `Functional Composition`: hỗ trợ việc kết hợp các `imperative` và `reactive` functions. - Hỗ trợ `REST` trong việc hiển thị các `Function` dưới dạng `Endpoint`. - `Streaming data` (thông qua Apache Kafka, RabbitMQ, Solace) đến/từ các `Function` thông qua `Spring Cloud Stream`. - Đóng gói các `Function` cho quá trình `deployment`, dành riêng cho nền tảng đích (như `AWS Lambda`, `Azure Functions`, `Google Cloud Functions`, hoặc các nhà cung cấp không có máy chủ). #### 1.3.4. Spring Cloud Stream `Spring Cloud Stream` là 1 framework được thiết kế để giúp xây dựng các ứng dụng `event-driven`, `streaming` và có thể mở rộng. Nguyên tắc cốt lõi của nó là giúp nhà phát triển tập trung việc xây dựng `logic code` còn phần cơ sở hạ tâng như tích hợp với `message broker` sẽ được `Spring Cloud Stream` đảm nhiệm. `Spring Cloud Stream` tận dụng khả năng thích ứng với từng nhà cung cấp `message broker` khác nhau, đồng thời cung cấp các lớp trừu tượng để đảm bảo việc trải nghiệm là nhất quán bất kể nền tảng phần mềm nào. Chỉ cần thêm các `dependency` của các `message broker` và chỉ cần cấu hình trong `application.yml` là có thể chuyển đổi giữa các `message broker` một cách dễ dàng mà không cần động chạm vào `code`. Framework hỗ trợ tích hợp với RabbitMQ, Apache Kafka, Kafka Streams và Amazon Kinesis. Ngoài ra còn có các tiện ích tích hợp do đối tác duy trì dành cho Google PubSub, Solace PubSub+, Azure Event Hubs và Apache RocketMQ. Các khối xây dựng cốt lõi của `Spring Cloud Stream`: - `Destination Binders`: là các thành phần chịu trách nhiệm cung cấp sự tích hợp với các hệ thống messaging bên ngoài. - `Destination Bindings`: là cầu nối giữa hệ thống messaging bên ngoài và application code (producer/consumer) do end user cung cấp. - `Messages`: Cấu trúc dữ liệu chuẩn được producer và consumer sử dụng để liên lạc với `Destination Binder` (và do đó, các ứng dụng khác thông qua hệ thống messaging bên ngoài). `Spring Cloud Stream` trang bị cho ứng dụng Spring Boot một `Destination binder` tích hợp liền mạch với hệ thống messaging bên ngoài. Chất kết dính này đảm nhận trách nhiệm thiết lập các kênh liên lạc giữa producer và consumer của ứng dụng cũng như các thực thể trong hệ thống nhắn tin (chẳng hạn như `exchange` và `queue` trong RabbitMQ). Các kênh liên lạc này, được gọi là `Destination Bindings`, đóng vai trò kết nối giữa ứng dụng và nhà môi giới (message broker). `Destination Bindings` có thể hoạt động như kênh đầu vào hoặc kênh đầu ra. Theo mặc định, `Spring Cloud Stream` ánh xạ từng liên kết, cả đầu vào và đầu ra, tới một `exchange` trong RabbitMQ (cụ thể là trao đổi chủ đề). Ngoài ra, đối với mỗi liên kết đầu vào, nó liên kết một `queue` với sàn giao dịch liên quan. Hàng đợi này đóng vai trò là nguồn mà từ đó người tiêu dùng nhận và xử lý các sự kiện. Cấu hình này cung cấp cơ sở hạ tầng cần thiết để triển khai kiến trúc hướng sự kiện dựa trên mô hình pub/sub. #### 1.3.5. Apache Kafka `Apache Kafka` là một hệ thống `messaging` mã nguồn mở được thiết kế để xử lý `streaming data` trên các ứng dụng `real-time`, `thông lượng cao` và `scalable`. Được sử dụng để xây dựng ứng dụng `real-time` và `streaming data pipeline` và các ứng dụng mà phù hợp với `data streams`. Một số thành phần chính của `Apache Kafka`: - `Producers`: Chịu trách nhiệm cho việc gửi, publish `messages` tới `Kafka topics`. Chúng gửi `messages` tới `Kafka topics` cụ thể và `Kafka` sẽ thêm các `messages` này vào `Topic's Log`. - `Topics`: `Kafka` tổ chức các dữ liệu vào trong `topics`. Một `topics` là một một luồng dữ liệu tham gia, nó có thể được chia thành nhiều `partitions`. Mỗi `message` trong `topic` có một `offset` (vị trí) duy nhất. - `Brokers`: Là các máy chủ `Kafka`, nó quản lý các `storage (lưu trữ)` và `replication (bản sao)` của `topics`. Chúng có trách nhiệm cho việc nhận `messages` từ `producers`, gán `offset` cho `messages` và phân phối `messages` tới `consumers`. - `Partitions`: Một `topic` có thể được chia thành nhiều `partitions`, cho phép cho các quá trình xử lý song song và cân bằng tải. Mỗi `partition` là một chuỗi `messages` có `thứ tự, không thể thay đổi`, và mỗi `message` trong `partition` có một `offset` duy nhất. - `Offsets`: Là một số duy nhất được gán cho mỗi `message` trong `partition`. `Offset` được sử dụng để theo dõi tiến trình của các `consumers`. `Consumers` có thể điều khiển `offset` của mình, cho phép chúng tua lại hoặc nhảy qua các `messages` nếu cần. - `Replication`: `Kafka` cho phép các `topic` được nhân rộng trên khắp các `brokers` để đảm bảo khả năng chịu lỗi. Sự nhân rộng cung cấp dữ liệu dư thừa để cho phép chuyển đổi dự phòng và tính sẵn sàng cao. - `Consumers`: Đọc `messages` từ `topics`. Chúng đăng ký `subcribe` tới một hoặc nhiều `topics` và xử lý `messages` được gửi tới chúng bằng các `partitions` cụ thể. Mỗi `consumer` duy trì `offset` của mình, cho phép chúng theo dõi tiến trình của mình ở `topic` mà chúng đang đọc. - `Consumer Groups`: Các `consumers` có thể tổ chức thành `consumer groups`. Mỗi `message` được gửi tới một `topic` sẽ đchỉ được gửi tới 1 `consumer` trong mỗi `consumer group`. Điều này cho phép tiến trình xử lý song song các `messages` trên nhiều `consumer`. - `Streams`: `Kafka Streams` là một thư viện Java được tích hợp sẵn trong `Kafka` cho phép xử lý `streaming data` trên `Kafka`. Nó cho phép xây dựng `consumer`, `producer`, `transform`, và sản xuất ra dữ liệu trong `thời gian thực`. Tiến tình của `Kafka` bên phía `Producer`: 1. `Producer Configurations`: Trước khi đẩy `message` vào `Kafka`, một `producer` cần phải được cấu hình. Điều này liên quan đến việc thiết lập các `thuộc tính` như `địa chỉ Kafka broker`, `serialization format cho message` và các cấu hình tùy chọn khác như `nén` hoặc `batching (nhóm)`. 2. `Topic Selection`: `Producer` cần chọn `topic` mà `message` sẽ được gửi tới. `Topics` là các luồng dữ liệu được xác định trước trong `Kafka`. Nếu `topic` không tồn tại, nó sẽ được tạo ra tự động, phụ thuộc vào cấu hình của `broker`. 3.`Message Production`: `Producer` gửi `message` tới `Kafka` bằng cách sử dụng API thư viện `Kafka client`. `Produce` chỉ định đích đến `topic` và `serialized message`. Nó có thê cung cấp một `partition key (optional)` để kiểm soát phân vùng nào `message` sẽ được ghi vào. 4. `Partition Assignment`: Nếu một `partition key` được cung cấp, `Kafka` sẽ sử dụng nó để xác định `partition` mà `message` sẽ được ghi vào. Nếu không, `Kafka` sẽ sử dụng thuật toán `round-robin` hoặc `hashing` để phân vùng `message` đồng đều qua các `partition`. 5. `Message Routing & offset assignment`: `Producer` gửi tin nhắn đến `Kafka broker` thích hợp dựa trên `topic` đích và `partition` được chỉ định cho `message`. `Broker` nhận được tin nhắn và thêm chúng vào log của `partition` tương ứng một cách lâu dài và có trật tự với sự trợ giúp của `offset id`. 6. `Message Replication`: `Kafka` đảm bảo tính sẵn sàng cao và khả năng chịu lỗi bằng việc nhân bản `message` trên nhiều `broker`. Mỗi `message` được ghi vào `partition` chính, `Kafka` sẽ sao chép không đồng bộ chúng sang các bản sao khác của `partition`. 7. `Acknowledgement and Error Handling`: `Producer` nhận được `acknowledgement` từ `Kafka` sau khi `message` được ghi thành công vào `partition` chính. `Producer` có thể xử lý bất kỳ lỗi, thử lại, lỗi tiềm ẩn nào dựa trên `acknowledgement` nhận được. Phụ thuộc vào chế độ `acknowledgement` được cấu hình, `producer` có thể chờ đợi `acknowledgement` từ tất cả các `bản sao` hoặc chỉ cần `bản sao chính`. Tiến trình của `Kafka` bên phía `Consumer`: 1. `Consumer Group and Topic Subscription`: Các `Consumer` trong `Kafka` thường được tổ chức thành các `consumer group`. Trước khi đọc `message`, `consumer` cần tham gia vào một `consumer group` và đăng ký một hoặc nhiều `topic` mà nó muốn đọc `message` từ đó. 2. `Partition Assignment`: `Kafka` chỉ định các `partition` của các `topic` đến các `consumer group`. Mỗi `partition` được sử dụng bởi chỉ một `consumer` trong `consumer group`. `Kafka` đảm bảo sự phân bổ công bằng của `partition` giữa các `consumer` để đạt được quá trình xử lý song song. 3. `Offset Management`: Mỗi `consumer` duy trì `offset` của mình cho mỗi `partition` mà nó đọc `message`. ban đầu, `offset` được đặt ở cuối `commited offset` hoặc ở `staring offset` được chỉ định. Khi `consumer` đọc `message`, `offset` sẽ được cập nhật để theo dõi tiến trình đọc `message`. 4. `Fetch Request`: `Consumer` gửi `fetch request` tới `Kafka broker(s)` mà nó được kết nối. `Fetch request` chứa `topic`, `partition`, và `offset` mà `consumer` muốn đọc `message` từ đó. `Request` cũng chỉ định số lượng tối đa của `message` được tìm nạp trong mỗi `fetch request`. 5. `Message Fetching`: Khi nhận được `fetch request`, `Kafka broker` sẽ lấy các `message` được yêu cầu từ `parittion's log` tương ứng. Nó sẽ trả về `message` cho `consumer` trong `fetch response`. `Response` chứa `message`, `offset`, và `metadata` khác. 6. `Message Processing`: Khi `Consumer` nhận được `message`, nó sẽ xử lý `message` theo logic của mình. Quá trình xử lý này có thể bao gồm các phép biến đổi `transformation`, tổng hợp `aggregation`, tính toán `calculation`, hoặc các hoạt động khác dựa trên nghiệp vụ yêu cầu. 7. `Committing the Offset`: Sau khi xử lý chuỗi `message`, `consumer` cần cập nhật `offset` của mình tới `Kafka broker`. Điều này chỉ ra rằng `consumer` đã hoàn thành xử lý `message` có `offset id` đó. `Committing offset` đảm bảo tiến trình của `consumer` được `duy trì và có thể được tiếp tục` từ điểm đó trong trường hợp thất bại hoặc khởi động lại. ## 2. Xây dựng `Message Microservices` với `Spring Cloud Function`. ### 2.1. Cài đặt thư viện. Để sử dụng `Spring Cloud Function`, chúng ta cần thêm các dependency sau vào file `pom.xml`: 1. `spring-cloud.version`: phiên bản của `Spring Cloud` vào `properties` của `pom.xml`. 2. Bộ quản lý dependency của `Spring Cloud` vào `dependencyManagement` của `pom.xml`. 3. `spring-cloud-function-context`: thư viện chính của `Spring Cloud Function`. >pom.xml ```xml org.springframework.cloud spring-cloud-function-context ``` 4. (OPTIONAL) `spring-cloud-starter-function-web`: thư viện hỗ trợ việc xây dựng `RESTful API` cho các `Function`, `Supplier`, `Consumer`. Có thể tùy chọn thêm vào `pom.xml` nếu cần show các `Function` dưới dạng `Endpoint`. Ví dụ tên `Function` là `uppercase`, khi thêm thư viện này, chúng ta có thể truy cập `uppercase` thông qua `http://localhost:8080/uppercase`. Tuy nhiên không cần thiết phải thêm thư viện này nếu không cần show các `Function` dưới dạng `Endpoint`. >pom.xml ```xml org.springframework.cloud spring-cloud-starter-function-web ``` ### 2.2. Xây dựng `Function`. 1. Đầu tiên chúng ta cần tạo ra 1 `DTO` class để định nghĩa dữ liệu đầu vào và dữ liệu đầu ra của `Function`. Dùng để giao tiếp với `Message Broker`. >dto/AccountsMsgDto.java ```java ... public record AccountsMsgDto(Long accountNumber, String name, String email, String mobileNumber) {} ``` 2. Tiếp theo, chúng ta sẽ tạo ra các `@Bean` `Function` trong một class được đánh dấu là `@Configuration` để `Spring` có thể quét và tạo ra các `Bean` từ `Function` để xử lý dữ liệu. Các `logical` xử lý dữ liệu sẽ được viết bằng `Lambda Expression`. Có các loại `FunctionalInterface` sau: - `Function`: nhận vào `T` và trả về `R`: Hoạt động với Method `POST`. `...return T -> { return R; }` - `Supplier`: không nhận vào và trả về `R`: Hoạt động với Method `GET`. `...return () -> { return R; }` - `Consumer`: nhận vào `T` và không trả về: Hoạt động với Method `POST`. `... return (T) -> { ... }` >functions/MessageFunctions.java ```java import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration // Đánh dấu lớp này là một lớp cấu hình Spring, tạo ra các bean public class MessageFunctions { private final Logger logger = LoggerFactory.getLogger(getClass()); @Bean // Đánh dấu phương thức này là một bean public Function email() { // Tạo ra một bean kiểu Function với đầu vào và đầu ra là AccountsMsgDto, endpoint là /email return accountsMsgDto -> { // accountsMsgDto là đầu vào của hàm logger.info("Sending email to: " + accountsMsgDto.toString()); return accountsMsgDto; // Trả về đầu ra của hàm }; // Phải được viết bằng Lambda Expression (->) } @Bean public Function sms() { // Tạo ra một bean kiểu Function với đầu vào là AccountsMsgDto và đầu ra là Long, endpoint là /sms return accountsMsgDto -> { // accountsMsgDto là đầu vào của hàm logger.info("Sending SMS to: " + accountsMsgDto.toString()); return accountsMsgDto.accountNumber(); // Trả về đầu ra của hàm }; // Phải được viết bằng Lambda Expression (->) } @Bean public Supplier name() { // Tạo ra một bean kiểu Supplier với đầu ra là String, endpoint là /name return () -> { // Đầu vào của hàm là rỗng logger.info("Getting name example"); return "John Doe"; // Trả về đầu ra của hàm }; // Phải được viết bằng Lambda Expression (->) } @Bean public Consumer print() { // Tạo ra một bean kiểu Consumer với đầu vào là String, endpoint là /print return s -> logger.info("Printing: " + s); // s là đầu vào của hàm, và trong Lambda Expression (->) không có return } } ``` Nếu sử dụng thư viện `spring-cloud-starter-function-web`, chúng ta có thể truy cập các `Function` thông qua `Endpoint`: - `http://localhost:8080/email` - `http://localhost:8080/sms` - `http://localhost:8080/name` - `http://localhost:8080/print` Ngoài ra chúng ta có thể gộp 2 hoặc nhiều `Function` thành một `Function` lớn bằng cách cấu hình trong `application.yml`, các tên `Function` sẽ được phân cách bằng dấu `|`. Kết quả trả về sẽ là kết qua của `function` cuối cùng trong chuỗi `Function`. > Lúc này, endpoint của `Function` lớn sẽ là gộp tên của các `Function` thành một chuỗi, ví dụ `/emailsms` hoặc `/smsemail` hoặc `/emailsmssmsemail`. Và đặc biệt nó sẽ thực hiện `logical` của các `Function` cùng một lúc. >application.yml ```yml spring: cloud: function: definition: email|sms # Gộp 2 Function email và sms thành một Function lớn, endpoint là /emailsms hoặc /smsemail hoặc /emailsmssmsemail.... đồng thời thực hiện logical của cả 2 Function email và sms ``` Để định nghĩa các `Function` độc lập, ta ngăn cách các tên `Function` bằng dấu `;` >application.yml ```yml spring: cloud: function: definition: email;sms # Định nghĩa 2 Function email và sms độc lập, endpoint là /email và /sms ``` > LƯU Ý: MẶC DÙ KHÔNG CẦN ĐỊNH NGHĨA `spring.cloud.function.definition` TRONG `application.yml`, NHƯNG NẾU KHÔNG ĐỊNH NGHĨA, KHI SỬ DỤNG VỚI `SPRING CLOUD STREAM` SẼ GẶP LỖI VÌ KHÔNG TÌM THẤY `FUNCTION` ĐỂ XỬ LÝ DỮ LIỆU. CHO NÊN CẦN PHẢI ĐỊNH NGHĨA TẤT CẢ CÁC `FUNCTION` CẦN SỬ DỤNG TRONG `application.yml`. ## 3. Xây dựng `Message Microservices` với `Spring Cloud Stream` + `Spring Cloud Function` + `RabbitMQ`. ### 3.1. Tạo `image` cho `RabbitMQ`. Chúng ta sẽ sử dụng `Docker` để tạo `image` cho `RabbitMQ`. ```bash docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.13-management ``` Trong đó, port `5672` là port mặc định để giao tiếp với `RabbitMQ`, port `15672` là port để truy cập giao diện `RabbitMQ Management`. ### 3.2. Cài đặt thư viện, dependency. Vì trong `Spring Cloud Stream` đã có sẵn `spring-cloud-function`, nên chúng ta chỉ cần thêm dependency của `Spring Cloud Stream` và `spring-cloud-stream-binder-rabbit` để kết nối với `RabbitMQ`. > LƯU Ý: CẦN PHẢI CÀI ĐẶT CÁC DEPENDENCY NÀY Ở CÁC SERVICE CẦN SỬ DỤNG `SPRING CLOUD STREAM` VÀ `RABBITMQ` ĐÊ GIÚP SERVICE CÓ THỂ GỬI VÀ NHẬN DỮ LIỆU QUA `RABBITMQ`. 1. `spring-cloud-stream`: thư viện chính của `Spring Cloud Stream`. >pom.xml ```xml org.springframework.cloud spring-cloud-stream ``` 2. `spring-cloud-stream-binder-rabbit`: thư viện kết nối `Spring Cloud Stream` với `RabbitMQ`. >pom.xml ```xml org.springframework.cloud spring-cloud-stream-binder-rabbit ``` 3. Dependemcy phục vụ cho việc test `Spring Cloud Stream` với `RabbitMQ`. >pom.xml ```xml org.springframework.cloud spring-cloud-stream-test-binder test ``` ### 3.3. Cấu hình trong `application.yml`. #### 3.3.1. Cấu hình kết nối tới `RabbitMQ`. Cần cấu hình kết nối tới `RabbitMQ` trong `application.yml` của các service cần sử dụng `Spring Cloud Stream` và `RabbitMQ` để gửi và nhận dữ liệu. ```yml spring: rabbitmq: # Cấu hình kết nối tới RabbitMQ host: localhost # Host của RabbitMQ port: 5672 # Port của RabbitMQ mặc định là 5672 là cổng giaot tiếp với RabbitMQ username: guest # Tên đăng nhập mặc định vào RabbitMQ password: guest # Mật khẩu đăng nhập mặc định vào RabbitMQ connection-timeout: 10s # Thời gian kết nối tối đa với RabbitMQ ``` #### 3.3.2. Định nghĩa các `Function` cần sử dụng. 1. Cần định nghĩa các `Function` cần sử dụng trong `application.yml` của các service cần sử dụng `Spring Cloud Stream` và `RabbitMQ` để gửi và nhận dữ liệu. Trước hết, cần định nghĩa các `Function` trong `MessageFunctions.java` như đã hướng dẫn ở phần `###2.2`. >application.yml ```yml spring: cloud: function: definition: email|sms;name|print # Định nghĩa 2 Function email và sms độc lập, endpoint là /email và /sms ``` 2. Cấu hình `Spring Cloud Stream` để sử dụng `RabbitMQ` trong `application.yml` của các service cần sử dụng `Spring Cloud Stream` và `RabbitMQ` để gửi và nhận dữ liệu. - Với luồng vào `in` là luồng kết nối giữa `Queue` và `Function` (đọc dữ liệu từ `Queue` và xử lý dữ liệu). - Với luồng ra `out` là luồng kết nối giữa `Function` và `exchange` (gửi dữ liệu từ `Function` tới `exchange`). > LƯU Ý: NÊN ĐẶT TÊN CHO CÁC `BINDINGS` THEO CÚ PHÁP `-in-`, VÍ DỤ `email-in-0`, `sms-in-0`, `name-in-0`, `print-in-0` ĐỂ DỄ DÀNG NHẬN BIẾT VÀ QUẢN LÝ. VÀ CÁC `BINDINGS` LÀ `IN` THÌ NÊN ĐỂ `GROUP` LÀ TÊN CỦA `APPLICATION` ĐỂ NHÓM CÁC `BINDINGS` CÙNG MỘT NHÓM. >application.yml ```yml spring: cloud: function: definition: email|sms;name;print # Định nghĩa 2 Function email và sms độc lập, endpoint là /emailsms, /name, /print stream: bindings: emailsms-in-0: # Đặt tên cho Function email và sms, theo cú pháp -in- destination: send-message # Đặt tên cho Queue, tên này sẽ được sử dụng trong RabbitMQ như là 1 tên của channel group: ${spring.application.name} # Tên của nhóm, nên sử dụng với các binding `in` emailsms-out-0: # Đặt tên cho Function email và sms, theo cú pháp -out- destination: message-sent # Đặt tên cho Exchange, tên này sẽ được sử dụng trong RabbitMQ như là 1 tên của channel ``` > VỚI CÁC TÊN `BINDINGS OUT` KHÔNG ĐƯỢC ĐỊNH NGHĨA TRONG `application.yml` HOẶC KHÔNG CÓ TÊN TRÙNG VỚI TÊN CỦA `functions` nào trong `SPRING CLOUD FUNCTION`, THÌ SỬ `INJECT` CLASS `STREAMBRIDGE` VÀ SỬ DỤNG PHƯƠNG THỨC `send` ĐỂ GỬI DỮ LIỆU TỚI `EXCHANGE` MONG MUỐN. VÍ DỤ 1. Message Service: Nhận dữ liệu từ `Queue` `send-message` và gửi dữ liệu tới `Exchange` `message-sent`. >`Message Service`: MessageFunctions.java - Định nghĩa các Function để xử lý dữ liệu trên `Queue` và gửi dữ liệu tới `Exchange`. ```java @Configuration // Đánh dấu lớp này là một lớp cấu hình Spring, tạo ra các bean public class MessageFunctions { private final Logger logger = LoggerFactory.getLogger(getClass()); @Bean // Đánh dấu phương thức này là một bean public Function email() { // Tạo ra một bean kiểu Function với đầu vào và đầu ra là AccountsMsgDto, endpoint là /email return accountsMsgDto -> { // accountsMsgDto là đầu vào của hàm logger.info("Sending email to: {}", accountsMsgDto.toString()); return accountsMsgDto; // Trả về đầu ra của hàm }; // Phải được viết bằng Lambda Expression (->) } @Bean public Function sms() { // Tạo ra một bean kiểu Function với đầu vào là AccountsMsgDto và đầu ra là Long, endpoint là /sms return accountsMsgDto -> { // accountsMsgDto là đầu vào của hàm logger.info("Sending SMS to: {}", accountsMsgDto.toString()); return accountsMsgDto.accountNumber(); // Trả về đầu ra của hàm }; // Phải được viết bằng Lambda Expression (->) } } ``` >`Message Service`: application.yml - Cấu hình kết nối tới `RabbitMQ` và `Spring Cloud Stream`. ```yml server: port: 9010 spring: application: name: "messages" cloud: function: definition: email|sms # Gop 2 function email va sms thanh 1 function voi entrypoint la emailsms, no se goi 2 function email va sms cung 1 luc stream: # Khai bao cac binding giua cac function bindings: emailsms-in-0: # Khai bao binding voi quy tac ten la -in- (in la input giao tiep voi queue do service khac gui den) destination: send-message # Ten queue ma service nay se nhan message tu service khac group: ${spring.application.name} # Group cua service, dung de phan biet cac service khac nhau emailsms-out-0: # Khai bao binding voi quy tac ten la -out- (out la output giao tiep voi exchange de gui message den service khac) destination: message-sent # Ten exchange ma service nay se gui message den service khac rabbitmq: host: localhost username: guest password: guest port: 5672 connection-timeout: 10s ``` 2. Account Service: Gửi dữ liệu tới `Queue` `send-message` và nhận dữ liệu từ `Exchange` `message-sent`. >`Account Service`: AccountFunctions.java - Định nghĩa các Function để gửi dữ liệu tới `Queue` và nhận dữ liệu từ `Exchange`. ```java package com.didan.microservices.accounts.functions; import com.didan.microservices.accounts.service.IAccountsService; import java.util.function.Consumer; import java.util.function.Supplier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class AccountsFunction { private final Logger logger = LoggerFactory.getLogger(getClass()); @Bean public Consumer updateCommunication(IAccountsService accountsService) { // Consumer là một functional interface, nó có một phương thức accept() nhận vào một tham số Long và không trả về giá trị return accountNumber -> { // accountNumber là tham số đầu vào của hàm logger.info("Updating Communication status for the account number: {}", accountNumber.toString()); accountsService.updateCommunicationStatus(accountNumber); // Gọi phương thức updateCommunicationStatus() của accountsService để thực hiện logic code }; } } ``` > `Account Service`: AccountsServiceImpl.java - Nơi sử dụng `StreamBridge` để gửi dữ liệu tới `Exchange`, nếu không định nghĩa `Functions` trong `AccountFunctions.java`. ```java @Service @AllArgsConstructor public class AccountServiceImpl implements IAccountsService { private final Logger logger = LoggerFactory.getLogger(getClass()); private final StreamBridge streamBridge; ... private void sendToExchange(Accounts accounts, Customer customer) { var accountsMsgDto = new AccountsMsgDto(accounts.getAccountNumber(), customer.getName(), customer.getEmail(), customer.getMobile()); // Tạo ra một AccountsMsgDto mới với các thông tin từ accounts và customer, Lưu ý kiểu dữ liệu gửi đi phải trùng với dữ liệu mà các Function đang lắng nghe logger.info("Sending Communication request for the details: {}", accountsMsgDto); var result = streamBridge.send("sendCommunication-out-0", accountsMsgDto); // Gửi thông tin đến exchange với tên là sendCommunication-out-0 đã được định nghĩa trong application.properties logger.info("Is the Communication request successfully triggered ? : {}", result); } } ``` >`Account Service`: application.yml - Cấu hình kết nối tới `RabbitMQ` và `Spring Cloud Stream`. ```yml spring: cloud: function: definition: updateCommunication # Đinh nghia function updateCommunication, voi ten function la updateCommunication da co trong AccountFunction.java stream: bindings: # Khai bao cac binding giua cac function sendCommunication-out-0: # Khai bao binding voi quy tac ten la -out- (out la output giao tiep voi exchange de gui message den service khac) destination: send-message # Ten exchange ma service nay se gui message den service khac updateCommunication-in-0: # Khai bao binding voi quy tac ten la -in- (in la input giao tiep voi queue do service khac gui den) destination: message-sent # Ten queue ma service nay se nhan message tu service khac group: ${spring.application.name} # Group cua service, dung de phan biet cac service khac nhau rabbitmq: host: localhost port: 5672 username: guest password: guest connection-timeout: 10s ``` SAU KHI CẤU HÌNH XONG, CHÚNG TA CẦN CHẠY CÁC SERVICE THEO THỨ TỰ SAU: `RABBITMQ`, `MESSAGE SERVICE`, `ACCOUNT SERVICE` ĐỂ KIỂM TRA XEM CÁC SERVICE CÓ THỂ GỬI VÀ NHẬN DỮ LIỆU QUA `RABBITMQ` ĐƯỢC KHÔNG. KHI ĐĂNG KÝ 1 USER MỚI TRÊN `ACCOUNT SERVICE`: 1. `ACCOUNT SERVICE` SẼ GỬI DỮ LIỆU TỚI `MESSAGE SERVICE` QUA EXCHANGE `send-message` BẰNG `STREAMBRIDGE` ĐỂ GỬI CÁC THỒNG TIN CỦA CLASS `AccountsMsgDto` ĐẾN `MESSAGE SERVICE` ĐỂ XỬ LÝ. 2. `MESSAGE SERVICE` NHẬN DỮ LIỆU TỪ `ACCOUNT SERVICE` QUA QUEUE `send-message` VÀ XỬ LÝ DỮ LIỆU THEO CÁC `FUNCTION` ĐÃ ĐƯỢC ĐỊNH NGHĨA TRONG `MessageFunctions.java`. VỚI KIỂU DỮ LIỆU ĐẦU VÀO GIỐNG VỚI KIỂU DỮ LIỆU MÀ `ACCOUNT SERVICE` GỬI ĐI. 3. SAU KHI XỬ LÝ XONG, `MESSAGE SERVICE` SẼ GỬI DỮ LIỆU TỚI `ACCOUNT SERVICE` QUA EXCHANGE `message-sent` ĐỂ `ACCOUNT SERVICE` VỚI KIỂU DỮ LIỆU LÀ `Long` ĐỂ CẬP NHẬT TRẠNG THÁI CỦA ACCOUNT ĐÃ ĐƯỢC GỬI TỚI `MESSAGE SERVICE`. 4. `ACCOUNT SERVICE` NHẬN DỮ LIỆU TỪ `MESSAGE SERVICE` QUA QUEUE `message-sent` VÀ XỬ LÝ DỮ LIỆU THEO CÁC `FUNCTION` ĐÃ ĐƯỢC ĐỊNH NGHĨA TRONG `AccountFunctions.java`. VỚI KIỂU DỮ LIỆU ĐẦU VÀO GIỐNG VỚI KIỂU DỮ LIỆU MÀ `MESSAGE SERVICE` GỬI ĐI. ## 4. Tạo file `Docker Compose` để chạy `Spring Cloud Function` + `Spring Cloud Stream` + `RabbitMQ`. Cần tạo 1 image `rabbitmq` sau đó cho `image` này join chung với các `image` của các service cần chạy trong `Docker Compose`. Cuối cùng để các `service` cần chạy với `event-driven` phụ thuộc vào `RabbitMQ`. >docker-compose.yml ```yml services: rabbitmq: image: rabbitmq:3.13-management # Sử dụng image RabbitMQ container_name: rabbitmq # Tên container hostname: rabbitmq # Hostname, thay vì localhost ports: - "5672:5672" # Port kết nối - "15672:15672" # Port quản lý healthcheck: # Kiểm tra kết nối tới RabbitMQ test: rabbitmq-diagnostics check_port_connectivity # Kiểm tra kết nối tới RabbitMQ interval: 10s # Thời gian kiểm tra timeout: 5s # Thời gian tối đa để kiểm tra retries: 10 # Số lần kiểm tra lại khi thất bại start_period: 5s # Thời gian bắt đầu kiểm tra sau khi Container khởi động networks: # Kết nối mạng - didan # Tên mạng kết nối với các service khác accounts: ... environtment: SPRING_RADIS_HOST: rabbitmq # Kết nối tới RabbitMQ networks: - didan messages: ... environtment: SPRING_RADIS_HOST: rabbitmq # Kết nối tới RabbitMQ networks: - didan networks: didan: # Tạo mạng kết nối các service driver: bridge # Sử dụng driver bridge ``` ## 6. So Sánh Apache Kafka vs RabbitMQ. `Kafta` và `RabbitMQ` đều là 2 hệ thống `message broker` phổ biến, được sử dụng rộng rãi trong việc xây dựng các ứng dụng `event-driven`, `streaming` và `microservices`. Dưới đây là một số điểm khác biệt giữa `Apache Kafka` và `RabbitMQ`: - `Design`: `Kafka` là một nền tảng `event streaming` một cách phân tán, trong khi `RabbitMQ` là một `message broker` - nhà môi giới tin nhắn. `Kafka` được thiết kế để xử lý các `dữ liệu lớn`, trong khi đó `RabbitMQ` được thiết kế để xử lý các `dữ liệu nhỏ` có nhiều phức tạp về yêu cầu định tuyến. - `Data retention`: `Kafka` lưu trữ dữ liệu trong `ổ đĩa`, trong khi `RabbitMQ` lưu trữ dữ liệu trong `Memory`. Vì thế `Kafka` có thể lưu trữ dữ liệu lớn hơn và lâu hơn so với `RabbitMQ`, trong khi đó `RabbitMQ` chỉ lưu trữ dữ liệu trong `Memory` nên đáp ứng với các yêu cầu về `độ trễ thấp`. - `Performance`: `Kafka` nhìn chung nhanh hơn so với `RabbitMQ` đặc biệt với `dữ liệu lớn`. Tuy nhiên `RabbitMQ` lại nhanh hơn với `Kafka` khi ứng dụng yêu cầu xử lý định tuyến phức tạp. - `Scalability`: `Kafka` có khả năng mở rộng tốt hơn so với `RabbitMQ` bởi vì `Kafka` có thể mở rộng theo chiều ngang để có thể thêm bất kỳ 1 `broker` nào vào `cluster`. ## 7. Cài đặt Kafka. Xem hướng dẫn cài đặt Kafka từ trang chủ: [Apache Kafka](https://kafka.apache.org/quickstart) ### 7.1. Cài đặt `Kafka` trên `Windows`, `Linux`, `MacOS`. 1. Tải file `kafka.tgz` từ trang tải [Kafka Download](https://kafka.apache.org/downloads). 2. Giải nén file `kafka.tgz` bằng lệnh `tar -xzf kafka.tgz`. 3. Di chuyển vào thư mục `kafka` bằng lệnh `cd kafka`. 4. Sử dụng `Kafka with KRaft` để chạy `Kafka` >Generate a Cluster UUID ```bash KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)" ``` >Format Log Directories ```bash bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties ``` >Start the Kafka Server ```bash bin/kafka-server-start.sh config/kraft/server.properties ``` ### 7.2. Cài đặt `Kafka` trên `Docker`. Để chạy `Kafka` trên `Docker`, chúng ta sử dụng `bitnami/kafka` image. Bởi vì `bitnami/kafka` image đã được cấu hình sẵn để chạy `Kafka` và `Zookeeper` trong môi trường `Docker`. ```bash docker run -d --name kafka-server --hostname kafka-server \ --network didan \ -e KAFKA_CFG_NODE_ID=0 \ -e KAFKA_CFG_PROCESS_ROLES=controller,broker \ -e KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \ -e KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \ -e KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka-server:9093 \ -e KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER \ -e KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://:9092 \ -e KAFKA_CFG_INTER_BROKER_LISTENER_NAME=PLAINTEXT \ bitnami/kafka:latest ``` Hoặc với `docker-compose.yml`: ```yml service: kafka: image: docker.io/bitnami/kafka:3.7 # Sử dụng image bitnami/kafka ports: - "9092:9092" # Port kết nối volumes: - ./kafka_data:/bitnami # Lưu trữ dữ liệu của Kafka trên ổ đĩa environment: # KRaft settings - KAFKA_CFG_NODE_ID=0 # Node ID của Kafka Server trong Cluster - KAFKA_CFG_PROCESS_ROLES=controller,broker # Process Roles của Kafka Server trong Cluster (controller, broker) - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093 # Quorum Voters của Kafka Server trong Cluster (@:9093), Nếu có nhiều Node thì thêm vào sau dấu phẩy - KAFKA_KRAFT_CLUSTER_ID=abcdefghijklmnopqrstuv # Cluster ID của Kafka Server trong Cluster, nên tạo ngẫu nhiên # Listeners - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 # Listener của Kafka Server (PLAINTEXT://:9092, CONTROLLER://:9093) - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://:9092 # Advertised Listener của Kafka Server (PLAINTEXT://:9092) - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT # Listener Security Protocol Map của Kafka Server (CONTROLLER:PLAINTEXT, PLAINTEXT:PLAINTEXT) - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER # Controller Listener Names của Kafka Server (CONTROLLER) - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=PLAINTEXT # Inter Broker Listener Name của Kafka Server (PLAINTEXT) # Clustering (Nếu muốn tạo nhiều Kafka Brokers thì thêm các cấu hình sau) - KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR=3 # Biểu thị Số lượng Replicas của Topic Offsets trong Kafka - KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=3 # Biểu thị Số lượng Replicas của Topic Transaction State Log trong Kafka - KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR=2 # Biểu thị Số lượng Replicas tối thiểu của Topic Transaction State Log trong Kafka networks: - didan # Kết nối mạng để join với các service khác ``` > LƯU Ý: CHÚNG TA CÓ THỂ TẠO NHIỀU `KAFKA BROKERS` BẰNG CÁCH THÊM CÁC `SERVICE` Y HỆT CHỈ THAY ĐỔI `NODE_ID`, THÊM VÀO `QUORUM_VOTERS` CÁC `NODE_ID` CỦA CÁC `KAFKA BROKERS` KHÁC, VÀ THÊM 3 CẤU HÌNH `OFFSET_TOPIC_REPLICATION_FACTOR`, `TRANSACTION_STATE_LOG_REPLICATION_FACTOR`, `TRANSACTION_STATE_LOG_MIN_ISR` ĐỂ TẠO RA 1 CLUSTER `KAFKA` VỚI NHIỀU `BROKERS`. ### 7.3. Kết hợp `Spring Cloud Stream` + `Spring Cloud Function` + `Kafka`. Do sử dụng `Spring Cloud Stream` và `Spring Cloud Function`, nên việc kết hợp `Kafka` với `Spring Cloud Stream` và `Spring Cloud Function` cũng tương tự như với `RabbitMQ`. Tất cả đều giồng như với `RabbitMQ`, chỉ cần thay đổi `dependency` và `configuration` trong `application.yml` để kết nối tới `Kafka` thay vì `RabbitMQ`. >pom.xml ```xml org.springframework.cloud spring-cloud-stream-binder-kafka ``` >application.yml ```yml spring: cloud: function: definition: email|sms;name;print # Định nghĩa 2 Function email và sms độc lập, endpoint là /emailsms, /name, /print stream: bindings: emailsms-in-0: # Đặt tên cho Function email và sms, theo cú pháp -in- destination: send-message # Tên của Topic, tên này sẽ được sử dụng trong Kafka như là 1 tên của channel group: ${spring.application.name} # Tên của nhóm, nên sử dụng với các binding `in` emailsms-out-0: # Đặt tên cho Function email và sms, theo cú pháp -out- destination: message-sent # Tên của Topic, tên này sẽ được sử dụng trong Kafka như là 1 tên của channel kafka: binder: brokers: localhost:9092 # Kết nối tới 1 hoặc nhiều Kafka Brokers ``` ### 7.4. Chạy `Docke Compose` với `Kafka`. Giống như phần `### 7.2`, sau khi viết service `kafka` xong, chúng ta thêm `environment` để kết nối tới `Kafka` trong các service cần sử dụng `Kafka`. ```yml ... accounts: ... environtment: SPRING_CLAOD_STREAM_KAFKA_BINDER_BROKERS: kafka:9092 # Kết nối tới Kafka networks: - didan messages: ... environtment: SPRING_CLAOD_STREAM_KAFKA_BINDER_BROKERS: kafka:9092 # Kết nối tới Kafka networks: - didan ... ```

Microservices Spring Boot: Event-driven trong hệ thống Microservices