Spring & Spring BootSpring Kafka là abstraction trên Apache Kafka client — KafkaTemplate để produce, @KafkaListener để consume.
xml
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>yaml
spring:
kafka:
bootstrap-servers: localhost:9092
consumer:
group-id: order-service
auto-offset-reset: earliest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.springframework.kafka.support.serializer.JsonSerializerProducer:
java
@Service
class OrderEventPublisher {
private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
public void publish(Order order) {
OrderEvent event = new OrderEvent(order.getId(), "ORDER_CREATED");
kafkaTemplate.send("order-events", order.getId().toString(), event)
.whenComplete((result, ex) -> {
if (ex != null) log.error("Failed to send", ex);
else log.info("Sent to partition {}", result.getRecordMetadata().partition());
});
}
}Consumer:
java
@Component
class OrderEventConsumer {
@KafkaListener(topics = "order-events", groupId = "inventory-service")
public void handle(OrderEvent event) {
inventoryService.reserve(event.getOrderId());
}
}Error handling: @RetryableTopic — retry tự động với backoff, Dead Letter Topic sau max retries.
Spring Kafka is an abstraction over the Apache Kafka client — KafkaTemplate for producing, @KafkaListener for consuming.
xml
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>yaml
spring:
kafka:
bootstrap-servers: localhost:9092
consumer:
group-id: order-service
auto-offset-reset: earliest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.springframework.kafka.support.serializer.JsonSerializerProducer:
java
@Service
class OrderEventPublisher {
private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
public void publish(Order order) {
OrderEvent event = new OrderEvent(order.getId(), "ORDER_CREATED");
kafkaTemplate.send("order-events", order.getId().toString(), event)
.whenComplete((result, ex) -> {
if (ex != null) log.error("Failed to send", ex);
else log.info("Sent to partition {}", result.getRecordMetadata().partition());
});
}
}Consumer:
java
@Component
class OrderEventConsumer {
@KafkaListener(topics = "order-events", groupId = "inventory-service")
public void handle(OrderEvent event) {
inventoryService.reserve(event.getOrderId());
}
}Error handling: @RetryableTopic — automatic retries with backoff, Dead Letter Topic after max retries.