introduzione

Apache Kafka è una piattaforma di streaming distribuita open source che gestisce lo streaming di dati in tempo reale su larga scala, ad alto throughput e con tolleranza ai guasti. Il sistema si basa su un modello di pubblicazione-sottoscrizione in cui i produttori pubblicano messaggi negli argomenti e i consumatori si iscrivono a tali argomenti per consumare i messaggi. Il caso d'uso di Apache Kafka è visto in scenari come analisi in tempo reale, architetture guidate da eventi, aggregazione di log, sistemi di messaggistica e costruzione di pipeline di dati scalabili.

Utilizzo di questo documento:

Puoi utilizzare questo documento per configurare e utilizzare Apache Kafka per produrre e consumare messaggi (contenuto JSON) nella tua applicazione di avvio Spring.

Prerequisiti

Hai bisogno di una buona conoscenza di Java, spring-boot, Apache Kafka, maven o Gradle prima di leggere questo documento, altrimenti ti suggerirò di controllare la loro documentazione e guida ufficiale.

Installazione

Prima di produrre messaggi per gli argomenti di Kafka, devi completare alcuni passaggi per configurare le dipendenze richieste nel tuo pom.xml (Maven) o build.gradle (progetto Gradle)

Per Maven:

For Gradle 
<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
   <groupId>org.springframework.kafka</groupId>
   <artifactId>spring-kafka</artifactId>
</dependency>

implementation ‘org.springframework.boot:spring-boot-starter’
implementation ‘org.springframework.kafka:spring-kafka’

Configurazione

Configurando le proprietà di Kafka nel file Application.properties, puoi anche avere le tue proprietà personalizzate. Specifica i server di bootstrap del server Kafka e qualsiasi proprietà di configurazione aggiuntiva necessaria, come l'ID del gruppo di consumatori.

Application.properties

spring.kafka.bootstrap-servers=<kafka-bootstrap-servers> 
spring.kafka.consumer.group-id=<consumer-group-id>
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.app.topic=<topic-name>

spring.kafka.bootstrap-servers=localhost:9092 
spring.kafka.consumer.group-id=group_category
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.app.topic=my-topic

Un producer Kafka è un componente che invia messaggi agli argomenti Kafka. Pubblica dati su Kafka, che uno o più consumatori Kafka possono quindi consumare.

Per creare un produttore Kafka, è necessario eseguire i seguenti passaggi:

1. Configurare le proprietà del producer Kafka: impostare le proprietà di configurazione necessarie per il producer Kafka, come i server bootstrap (indirizzi dei broker Kafka) e le impostazioni di serializzazione. Questo lo abbiamo già eseguito nell'ultimo passaggio.
2. Crea un produttore Kafka: per inviare messaggi a Kafka; puoi utilizzare il modello fornito da Spring Kafka. Ecco un esempio di un produttore semplice:

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Component
public class KafkaEventProducer {
 
 private static final Logger logger = LoggerFactory.getLogger(KafkaEventProducer.class);
 
 @Autowired
 private KafkaTemplate<String, String> kafkaTemplate;
 
 public void sendMessage(String topic, String message) {
   logger.info("Producing message [{}]", message);
   kafkaTemplate.send(topic, message);
 }
}

3. Configurare le proprietà del produttore di Kafka: configurare le proprietà di Kafka nel file application.properties. Specifica i server bootstrap del server Kafka e qualsiasi proprietà di configurazione aggiuntiva necessaria per il consumatore.
4. Crea un ascoltatore del consumatore Kafka: implementa un metodo nella tua applicazione che verrà chiamato ogni volta che viene ricevuto un nuovo messaggio da Kafka. Usa l'annotazione kafkaListener fornita da Spring Kafka.

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Component
public class KafkaEventConsumer {

private static final Logger logger = LoggerFactory.getLogger(KafkaEventConsumer.class);

@KafkaListener(topics = “<topic-name>”, groupId = “<consumer-group-id>”)
 public void consumeMessage(String message) {
  //Use log
 log.info("Consumed message [{}]", message);
 }
}

@KafkaListener(topics = “${spring.kafka.app.topic}”,groupId= “${spring.kafka.consumer.group-id}”)

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class MainApplication {

 @Autowired
 private KafkaEventProducer kafkaEventProducer;

 public static void main(String[] args) {
   SpringApplication.run(MainApplication.class, args);
 }
 
 public void YourMethod() {
 // Send a message using the Kafka producer
 kafkaEventProducer.sendMessage(“<topic-name>”, “Oh Kafka Boy How'z Everything?”);
 }
}

Conclusione

In conclusione, Spring Boot fornisce un eccellente supporto per l'integrazione di Apache Kafka nelle tue applicazioni. Con il supporto di Kafka di Spring Boot, puoi facilmente creare produttori e consumatori di Kafka, configurare le proprietà di Kafka e gestire l'elaborazione dei messaggi.
L'integrazione di Spring Boot con Kafka semplifica lo sviluppo di applicazioni basate su Kafka, fornendo un livello più elevato di astrazione e riducendo la quantità di codice boilerplate richiesto.

E prima di eseguire il codice, assicurati che il server Kafka sia in esecuzione e che gli argomenti siano stati creati.

Nota: ricordarsi di fare riferimento alla documentazione ufficiale di Spring Kafka ed esplorare gli esempi di Spring Kafka per comprendere meglio le varie funzionalità e opzioni disponibili per l'integrazione di Kafka in Spring Boot.

Grazie!