Apache Kafka is a distributed messaging system that provides a fast, scalable, and fault-tolerant way to publish and subscribe to streams of records. It is commonly used in modern applications for building real-time data pipelines and streaming applications.
To use Kafka in Java and Spring Boot applications, you need to include the Kafka client libraries in your project's dependencies. Here is an example of how to configure Kafka in a Spring Boot application:
1// Add the Kafka configuration properties to your application.properties
2spring.kafka.bootstrap-servers=localhost:9092
3spring.kafka.consumer.group-id=my-group
4
5// Create a Kafka consumer
6@Configuration
7@EnableKafka
8public class KafkaConsumerConfig {
9
10 @Value("${spring.kafka.bootstrap-servers}")
11 private String bootstrapServers;
12
13 @Value("${spring.kafka.consumer.group-id}")
14 private String groupId;
15
16 @Bean
17 public ConsumerFactory<String, String> consumerFactory() {
18 Map<String, Object> props = new HashMap<>();
19 props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
20 props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
21 props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
22 props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
23 return new DefaultKafkaConsumerFactory<>(props);
24 }
25
26 @Bean
27 public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
28 ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
29 factory.setConsumerFactory(consumerFactory());
30 return factory;
31 }
32
33 @KafkaListener(topics = "my-topic", groupId = "my-group")
34 public void listen(String message) {
35 // Process the received message
36 }
37
38}In this example, we define the Kafka consumer configuration using the @EnableKafka annotation. The ConsumerFactory and KafkaListenerContainerFactory beans are created, and the Kafka listener method listen processes the received message.
To use Kafka in a Java application without Spring Boot, you can use the Kafka Java client API directly. Here is an example of consuming messages from Kafka:
1import org.apache.kafka.clients.consumer.KafkaConsumer;
2import org.apache.kafka.clients.consumer.ConsumerRecords;
3import org.apache.kafka.clients.consumer.ConsumerRecord;
4import java.util.Properties;
5
6public class KafkaConsumerExample {
7
8 public static void main(String[] args) {
9 Properties props = new Properties();
10 props.put("bootstrap.servers", "localhost:9092");
11 props.put("group.id", "my-group");
12 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
13 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
14 KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
15 consumer.subscribe(Arrays.asList("my-topic"));
16 while (true) {
17 ConsumerRecords<String, String> records = consumer.poll(100);
18 for (ConsumerRecord<String, String> record : records) {
19 // Process the received record
20 }
21 }
22 consumer.close();
23 }
24
25}In this example, we create a Kafka consumer using the KafkaConsumer class from the Kafka Java client API. The consumer subscribes to the my-topic topic, and the received records are processed in the for loop.
Kafka provides various features such as message partitioning, replication, fault-tolerance, and scalability, making it a powerful messaging system for distributed applications. By integrating Kafka with Java and Spring Boot, you can build robust and efficient real-time data processing systems.
