Java Spring Boot: montando un productor y consumidor de Kafka • JnjSite.com

Esto es un codekata en Java con el framework de desarrollo Spring, en concreto usando la tecnología Spring Boot, para usar Apache Kafka.

Con Spring Boot conseguimos que el proyecto esté orientado a ser un microservicio, teniendo el servidor web Tomcat embebido. Por otro lado, mediante Kafka podemos dividir las aplicaciones en microservicios, comunicarnos entre sistemas con mensajes, convertir en asíncronos procesos pesados para recibir altos picos de peticiones procesando en background después, etcétera.. ¡vamos al grano!

Generando el proyecto desde cero

Como punto de partida tenemos que tener instalado Java, Maven y el Spring CLI. Y por supuesto, también Kafka. Si no tenemos el Spring CLI instalado entonces tendremos que generar el código fuente con alguna otra herramienta como por ejemplo la página web del Spring Boot initializer:
https://start.spring.io/

Si no tenemos Kafka instalado sólo hay que descargar los binarios y seguir las instrucciones desde:
https://kafka.apache.org/quickstart

Si tenemos el Spring CLI en nuestro sistema UNIX, es tan fácil como ir al terminal y escribir:

spring init -d=web,restdocs,kafka,kafka-streams,devtools testing-kafka

Esto nos generará el esqueleto para el proyecto bajo el directorio testing-kafka/. Así que ya podemos volver al terminal, y escribir:

cd testing-kafka
mvn spring-boot:run

..y si todo ha ido bien, se tiene que ver algo como lo siguiente:

Generando un productor y un consumidor de Kafka

Un caso base sin más complicaciones, con sólo lo mínimo para un productor de Kafka, podría ser algo como lo siguiente en un fichero src/main/java/com/example/testingkafka/component/KafkaMessageProducer.java:

package com.example.testingkafka.component;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class KafkaMessageProducer {
	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;

	@Value(value = "${topic.name}")
	private String topicName;

	public void sendMessage(String message) {
		kafkaTemplate.send(topicName, message);
	}
}

Para el consumidor de Kafka, sólo hace falta algo como lo siguiente, por ejemplo en un fichero consumidor lamado src/main/java/com/example/testingkafka/component/KafkaMessageConsumer.java:

package com.example.testingkafka.component;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaMessageConsumer {
    @KafkaListener(topics = "${topic.name}", groupId = "${topic.group.id}")
    public void listenTopic(String message) {
        System.out.println("Mensaje recibido: " + message);
    }
}

Unas configuraciones necesarias en el application.properties

Como ya empezamos a usar configuraciones en el productor y consumidor anteriores, entonces necesitamos dichas configuraciones en el fichero src/main/resources/application.properties:

topic.name=topic_pruebas
topic.group.id=group_id

Generando los mensajes desde un endpoint HTTP

Esto es lo que se llaman los controladores REST en Spring Boot. Por ejemplo, se puede hacer una clase dedicada a hacer unos puntos de entrada en un fichero src/main/java/com/example/testingkafka/rest/MainController.java:

package com.example.testingkafka.rest;

import com.example.testingkafka.component.KafkaMessageProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class MainController {
    @Autowired
    KafkaMessageProducer kafkaMessageProducer;

    @GetMapping("/pruebas")
    public String test() {
        kafkaMessageProducer.sendMessage("Prueba " + (int) (Math.random() * 1000 + 1));
        return "¡Producido mensaje en topic!";
    }
}

Este punto de entrada, simplemente recibirá peticiones en el path /pruebas, enviando mensajes a Kafka mediante el productor. A su vez, dicho mensaje producido en Kafka, lanzará el consumidor a continuación. El consumidor es como un listenner u observer, que estará a la espera de mensajes en Kafka consultando cada poco tiempo. Esta escucha del consumidor no es síncrona, es asíncrona, pero puede ser muy rápido.

Resumidas a cuentas, lo que hace esta estructura, es poder recibir muchas peticiones, las encola en Kafka, y las procesa asíncronamente con el consumidor.

Terminando y referencias

Si todo ha ido bien en el proceso, se tiene que poder ir a la URL http://localhost:8080/pruebas, en donde tenemos que ver algo parecido a lo siguiente en el navegador: