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: Thêm các thuộc tính cấu hình kết nối đến Kafka broker trong file application.properties:

    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 trong thư mục src/main/java/com/example/config:

    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<String, String> producerFactory() {
            Map<String, Object> 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<String, String> kafkaTemplate() {
            return new KafkaTemplate<>(producerFactory());
        }
    }
   

4. Tạo Kafka Producer Service: Tạo file KafkaProducerService.java trong thư mục src/main/java/com/example/service:

    package com.example.service;
   
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.stereotype.Service;
   
    @Service
    public class KafkaProducerService {
   
        private final KafkaTemplate<String, String> kafkaTemplate;
   
        public KafkaProducerService(KafkaTemplate<String, String> 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 trong thư mục src/main/java/com/example/controller:

    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:

       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:

       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:

       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, 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.