Sure! Below is an example of a simple Kafka producer and consumer using Spring Boot with the Spring Kafka library.
1. First, make sure you have the necessary dependencies in your `pom.xml` (Maven) or `build.gradle` (Gradle) file:
Maven:
```xml
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
</dependencies>
```
Gradle:
```gradle
dependencies {
implementation 'org.springframework.boot:spring-boot-starter-web'
implementation 'org.springframework.kafka:spring-kafka'
}
```
2. Create a Kafka configuration class to define the necessary Kafka properties:
```java
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 KafkaConfig {
private static final String KAFKA_BROKER = "localhost:9092";
private static final String TOPIC = "my-topic";
@Bean
public Map<String, Object> producerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_BROKER);
props.put(org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class);
props.put(org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class);
return props;
}
@Bean
public ProducerFactory<String, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(producerConfigs());
}
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
@Bean
public String topic() {
return TOPIC;
}
}
```
3. Create a Kafka producer to send messages:
```java
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
@Service
public class KafkaProducerService {
private final KafkaTemplate<String, String> kafkaTemplate;
private final String topic;
public KafkaProducerService(KafkaTemplate<String, String> kafkaTemplate, String topic) {
this.kafkaTemplate = kafkaTemplate;
this.topic = topic;
}
public void sendMessage(String message) {
kafkaTemplate.send(topic, message);
}
}
```
4. Create a Kafka consumer to receive messages:
```java
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class KafkaConsumerService {
@KafkaListener(topics = "my-topic", groupId = "my-group")
public void receiveMessage(String message) {
System.out.println("Received message: " + message);
}
}
```
5. Run the Spring Boot application:
```java
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class KafkaApplication {
public static void main(String[] args) {
SpringApplication.run(KafkaApplication.class, args);
}
}
```
Now, when you run the Spring Boot application, it will act as both a Kafka producer and consumer. When you send a message using the `KafkaProducerService.sendMessage()` method, the Kafka consumer (`KafkaConsumerService.receiveMessage()`) will receive and print the message.
Make sure you have Kafka running locally on the default port (9092) before running the Spring Boot application.
Please note that this example uses a simple setup with a single topic and group ID. In a real-world scenario, you may need more advanced configurations and error handling based on your use case.
