# Bài 10: Spring Boot Producer ### Mục tiêu Xây dựng một Producer để gửi thông điệp vào Kafka. ### Nội dung chính #### Cấu trúc Producer trong Spring Boot Trong Spring Boot, Producer chịu trách nhiệm gửi thông điệp vào Kafka. Producer thường được cấu hình và quản lý bởi Spring Kafka, sử dụng `KafkaTemplate` để gửi thông điệp một cách đơn giản và hiệu quả. #### Sử dụng KafkaTemplate để gửi thông điệp `KafkaTemplate` là thành phần chính được sử dụng để gửi thông điệp vào Kafka. Nó cung cấp các phương thức gửi thông điệp đồng bộ và không đồng bộ, cho phép xử lý phản hồi và lỗi một cách dễ dàng. ### Thực hành #### Viết một ứng dụng Spring Boot gửi thông điệp đến Kafka 1. **Tạo dự án Spring Boot**: Sử dụng Spring Initializr để tạo một dự án Spring Boot với các dependency cần thiết như `Spring Web` và `Spring Kafka`. 2. **Cấu hình** [`application.properties`](http://application.properties): Thêm các thuộc tính cấu hình kết nối đến Kafka broker trong file [`application.properties`](http://application.properties): ```java spring.kafka.bootstrap-servers=localhost:9092 spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer ``` 3. **Tạo lớp cấu hình Kafka Producer**: Tạo file [`KafkaProducerConfig.java`](http://KafkaProducerConfig.java) trong thư mục `src/main/java/com/example/config`: ```java package com.example.config; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.serialization.StringSerializer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import java.util.HashMap; import java.util.Map; @Configuration public class KafkaProducerConfig { @Bean public ProducerFactory producerFactory() { Map configProps = new HashMap<>(); configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); return new DefaultKafkaProducerFactory<>(configProps); } @Bean public KafkaTemplate kafkaTemplate() { return new KafkaTemplate<>(producerFactory()); } } ``` 4. **Tạo Kafka Producer Service**: Tạo file [`KafkaProducerService.java`](http://KafkaProducerService.java) trong thư mục `src/main/java/com/example/service`: ```java package com.example.service; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.stereotype.Service; @Service public class KafkaProducerService { private final KafkaTemplate kafkaTemplate; public KafkaProducerService(KafkaTemplate kafkaTemplate) { this.kafkaTemplate = kafkaTemplate; } public void sendMessage(String topic, String message) { kafkaTemplate.send(topic, message); } } ``` 5. **Tạo Controller để gửi thông điệp**: Tạo file [`KafkaController.java`](http://KafkaController.java) trong thư mục `src/main/java/com/example/controller`: ```java package com.example.controller; import com.example.service.KafkaProducerService; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; @RestController public class KafkaController { private final KafkaProducerService kafkaProducerService; public KafkaController(KafkaProducerService kafkaProducerService) { this.kafkaProducerService = kafkaProducerService; } @PostMapping("/send") public String sendMessage(@RequestParam String topic, @RequestParam String message) { kafkaProducerService.sendMessage(topic, message); return "Message sent!"; } } ``` 6. **Chạy ứng dụng Spring Boot**: * Chạy ứng dụng Spring Boot bằng cách sử dụng IDE hoặc dòng lệnh: ```sh mvn spring-boot:run ``` 7. **Kiểm tra gửi thông điệp**: * Sử dụng công cụ như Postman hoặc curl để gửi request đến endpoint `/send`: ```sh curl -X POST "http://localhost:8080/send?topic=my-topic&message=HelloKafka" ``` 8. **Xác minh thông điệp đã được gửi thành công**: * Mở một terminal mới và sử dụng Kafka Console Consumer để nhận message: ```sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning ``` * Bạn sẽ thấy message "HelloKafka" được hiển thị trên console. ### Câu hỏi củng cố kiến thức 1. **KafkaTemplate là gì và vai trò của nó trong Spring Kafka?** * **Trả lời**: `KafkaTemplate` là một thành phần chính trong Spring Kafka được sử dụng để gửi thông điệp tới Kafka. Nó cung cấp các phương thức để gửi thông điệp một cách đồng bộ và không đồng bộ, quản lý kết nối và xử lý lỗi. 2. **Làm thế nào để cấu hình kết nối Kafka trong Spring Boot?** * **Trả lời**: Bạn cần cấu hình các thuộc tính kết nối trong file [`application.properties`](http://application.properties), tạo các bean cấu hình cho `KafkaTemplate` và `ProducerFactory` trong một lớp cấu hình Kafka. 3. **Làm thế nào để kiểm tra rằng thông điệp đã được gửi thành công tới Kafka?** * **Trả lời**: Bạn có thể sử dụng Kafka Console Consumer để lắng nghe và nhận thông điệp từ topic mà bạn đã gửi thông điệp tới, sau đó kiểm tra xem thông điệp có xuất hiện trên console không. ### Kết luận Với bài viết này, bạn đã học cách xây dựng một Producer trong Spring Boot để gửi thông điệp tới Kafka. Bạn đã cấu hình các thuộc tính cần thiết, tạo các lớp cấu hình, và viết mã để gửi thông điệp. Bạn cũng đã thực hiện các bước kiểm tra để đảm bảo rằng thông điệp đã được gửi thành công. Trong các bài tiếp theo, chúng ta sẽ khám phá thêm về các khía cạnh khác của Spring Kafka, như Consumer và cách xử lý lỗi.