Ejemplo de consumidor de Spring Boot Kafka

Spring Boot es uno de los marcos más populares y más utilizados del lenguaje de programación Java. Es un marco basado en microservicios y hacer una aplicación lista para producción usando Spring Boot lleva mucho menos tiempo. Spring Boot facilita la creación de aplicaciones basadas en Spring independientes y de grado de producción que puede » simplemente ejecutar «. Entonces, algunas de las características principales de Spring Boot se enumeran a continuación.

• Cree aplicaciones Spring independientes
• Incruste Tomcat, Jetty o Undertow directamente.
• Proporcione dependencias «iniciales» para simplificar la configuración de compilación.
• Configure Spring y bibliotecas de terceros automáticamente siempre que sea posible.
• Proporcione funciones listas para producción, como comprobaciones de estado, métricas y configuración externalizada.
• Casi ninguna generación de código y ningún requisito para la configuración de XML.

Apache Kafka es un sistema de mensajería de publicación y suscripción. Un sistema de mensajería le permite enviar mensajes entre procesos, aplicaciones y servidores. En términos generales, Apache Kafka es un software donde los temas (un tema puede ser una categoría) se pueden definir y procesar. Las aplicaciones pueden conectarse a este sistema y transferir un mensaje al tema. Un mensaje puede incluir cualquier tipo de información, de cualquier evento en su blog personal o puede ser un mensaje de texto muy simple que desenstringría cualquier otro evento. Aquí discutiremos cómo podemos consumir mensajes de temas de Kafka y mostrarlos en nuestra consola con Spring Boot, donde Kafka es un requisito previo . 

Ejemplo:

Requisito previo : asegúrese de haber instalado Apache Kafka en su máquina local para lo cual debe saber ¿Cómo instalar y ejecutar Apache Kafka en Windows?

Paso 1: Vaya a este enlace y cree un proyecto Spring Boot. Agregue la dependencia » Spring for Apache Kafka » a su proyecto Spring Boot. 

Paso 2: cree un archivo de configuración denominado KafkaConfig . A continuación se muestra el código del archivo KafkaConfig.java .

// Java Program to Illustrate Kafka Configuration
  
package com.amiya.kafka.apachekafkaconsumer.config;
  
// Importing required classes
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.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;
  
// Annotations
@EnableKafka
@Configuration
  
// Class
public class KafkaConfig {
  
    @Bean
    public ConsumerFactory<String, String> consumerFactory()
    {
  
        // Creating a Map of string-object pairs
        Map<String, Object> config = new HashMap<>();
  
        // Adding the Configuration
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                   "127.0.0.1:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG,
                   "group_id");
        config.put(
            ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
            StringDeserializer.class);
        config.put(
            ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
            StringDeserializer.class);
  
        return new DefaultKafkaConsumerFactory<>(config);
    }
  
    // Creating a Listener
    public ConcurrentKafkaListenerContainerFactory
    concurrentKafkaListenerContainerFactory()
    {
        ConcurrentKafkaListenerContainerFactory<
            String, String> factory
            = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

Paso 3: cree un archivo de consumidor llamado KafkaConsumer

// Java Program to Illustrate Kafka Consumer
  
package com.amiya.kafka.apachekafkaconsumer.consumer;
  
// Importing required classes
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
  
@Component
  
// Class
public class KafkaConsumer {
  
    @KafkaListener(topics = "NewTopic",
                   groupId = "group_id")
  
    // Method
    public void
    consume(String message)
    {
        // Print statement
        System.out.println("message = " + message);
    }
}

Paso 4: ahora tenemos que hacer lo siguiente para consumir mensajes de temas de Kafka con Spring Boot

• Ejecute el servidor Apache Zookeeper
• Ejecute el servidor Apache Kafka
• Enviar los mensajes de Kafka Temas

Ejecute su servidor Apache Zookeeper usando este comando

C:\kafka>.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

De manera similar, ejecute su servidor Apache Kafka usando este comando

C:\kafka>.\bin\windows\kafka-server-start.bat .\config\server.properties

Ejecute el siguiente comando para enviar los mensajes de Kafka Topics

C:\kafka>.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic NewTopic

Paso 5: Ahora ejecute su aplicación Spring Boot. Asegúrese de haber cambiado el número de puerto en el archivo application.properties

server.port=8081

Ejecutemos la aplicación de arranque Spring dentro del archivo ApacheKafkaConsumerApplication

Salida: en la salida, puede ver cuando está enviando el mensaje desde Kafka Topics, se muestra en la consola en tiempo real.