Mình đã hướng dẫn các bạn cách gửi một message tới Apache Kafka server sử dụng class KafkaTemplate của Spring for Apache Kafka. Để consume các message từ các topics của Apache Kafka sử dụng Spring for Apache Kafka, các bạn có thể sử dụng annotation @KafkaListener. Cụ thể như thế nào? Hãy cùng nhau tìm hiểu trong bài viết này các bạn nhé!
Đầu tiên, mình sẽ tạo mới một Maven project để làm ví dụ:
Spring for Apache Kafka được khai báo như sau:
|
1 2 3 4 5 |
<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>3.3.1</version> </dependency> |
Để có thể consume message từ Apache Kafka server, đầu tiên, các bạn cần phải cấu hình thông tin bean của một class implement interface KafkaListenerContainerFactory. Mặc định, có 2 implementation chính cho interface KafkaListenerContainerFactory là ConcurrentKafkaListenerContainerFactory và RetryTopicListenerContainerFactoryDecorator. Thông thường chúng ta sẽ sử dụng class ConcurrentKafkaListenerContainerFactory các bạn nhé!
Class ConcurrentKafkaListenerContainerFactory sẽ sử dụng thông tin của Apache Kafka server, được cấu hình sử dụng class ConsumerFactory, để tạo listener đến các topics của Apache Kafka server. Cụ thể các bạn có thể cấu hình bean cho các class ConcurrentKafkaListenerContainerFactory và ConsumerFactory như sau:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
@Bean public ConsumerFactory<String, String> consumerFactory() { Map<String, Object> props = new HashMap<>(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); props.put(ConsumerConfig.GROUP_ID_CONFIG, "huongdanjava"); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); return new DefaultKafkaConsumerFactory<>(props); } @Bean public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); return factory; } |
Annotation @KafkaListener sẽ được sử dụng để annotate cho các phương thức sẽ consume các message từ Apache Kafka server.
Để Spring có thể process dựa vào annotation @KafkaListener này, các bạn cần phải khai báo thêm một annotation khác là @EnableKafka trong class cấu hình của Spring các bạn nhé. Cụ thể như sau:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 |
package com.huongdanjava.springapachekafka; import java.util.HashMap; import java.util.Map; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.common.serialization.StringDeserializer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.annotation.EnableKafka; import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; @Configuration @EnableKafka @ComponentScan public class AppConfiguration { @Bean public ConsumerFactory<String, String> consumerFactory() { Map<String, Object> props = new HashMap<>(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); return new DefaultKafkaConsumerFactory<>(props); } @Bean public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); return factory; } } |
Ở đây, mình cũng đã khai báo thêm annotation @ComponentScan để Spring có thể scan tất cả các class có chứa các phương thức được annotate với annotation @KafkaListener và tạo bean cho chúng.
Cho ví dụ của mình, mình có thể khai báo một class với annotation @Component có chứa phương thức listen() được annotate với annotation @KafkaListener như sau:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
package com.huongdanjava.springapachekafka; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component; @Component public class AppListener { @KafkaListener(topics = "huongdanjava", groupId = "huongdanjava") public void listen(String message) { System.out.println(message); } } |
Bây giờ, mình chỉ cần viết class main của ứng dụng để Spring tạo các bean dựa vào annotation sử dụng class AnnotationConfigApplicationContext như sau:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
package com.huongdanjava.springapachekafka; import org.springframework.context.annotation.AnnotationConfigApplicationContext; public class Application { public static void main(String[] args) throws InterruptedException { AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext( AppConfiguration.class); Thread.sleep(100000); } } |
thì khi chạy ứng dụng, nếu mình gửi một message “Hello World!” sử dụng ví dụ trong bài viết Gửi message tới Apache Kafka sử dụng class KafkaTemplate của Spring for Apache Kafka, các bạn sẽ thấy kết quả như sau:
Như vậy là chúng ta đã consume được message từ Apache Kafka server sử dụng Spring for Apache Kafka rồi đó các bạn!
