Tutoriel Kafka avec Java

package fr.tutoriel.kafka.dto;

/**
 * Exemple d'un objet 
 */
public class Person {
    private String firstName;
    private String lastName;
    private int age;    
    
    public Person() {

    }
    
    public Person(String firstName, String lastName, int age) {
        this.firstName = firstName;
        this.lastName = lastName;
        this.age = age;
    }
    
    public String getFirstName() {
        return firstName;
    }
    public void setFirstName(String msg) {
        this.firstName = msg;
    }
        
    public String getLastName() {
        return lastName;
    }
    
    public void setLastName(String name) {
        this.lastName = name;
    }

    public int getAge() {
        return age;
    }
    
    public void setAge(int age) {
        this.age = age;
    }

    @Override
    public String toString() {
        return     "{firstName:\""+firstName+"\", lastName:\""+lastName+"\", age:"+String.valueOf(age)+"}";        
    }
}

package fr.tutoriel.kafka.receiver;

import fr.tutoriel.kafka.dto.Person;

/**
 * Interface pour gerer les traitements
 *  
 * */
public interface IProcess {

        /**
         * Exécuter un traitement
         * @param info
         * @param person
         */
        public void execute(String info, Person person);
}

package fr.tutoriel.kafka.receiver;

import fr.tutoriel.kafka.dto.Person;

/**
 * Interface pour écouter la réception d'un message
 * 
 * Il serait possible d'utiliser la même interface pour écouter
 * un message avec une autre messagerie que Kafka (ex : RabbitMQ)
 * */
public interface IReceiver {
        public void listen(String topicName, Person person);
}

package fr.tutoriel.kafka.receiver;

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.beans.factory.annotation.Value;
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;
import org.springframework.kafka.support.serializer.JsonDeserializer;

import fr.tutoriel.kafka.dto.Person;

/**
 * Configuration du destinataire 
 */
@EnableKafka
@Configuration
public class KafkaReceiverConfig {

        @Value(value = "${kafka.bootstrapAddress:kafka:9092}") // Adresse du serveur Kafka pour envoyer les messages (il est possible de le configurer via application.properties)
    private String bootstrapAddress;
        
    private static final String groupId = "Tutorial"; // Définition du groupe ex: Tutorial
    
    @Bean
    public ConsumerFactory<String, Person> receiverFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapAddress); // Configuration de l'adresse du serveur
        props.put(ConsumerConfig.GROUP_ID_CONFIG,groupId);// Configuration du groupe
        return new DefaultKafkaConsumerFactory<>(props,new StringDeserializer(),new JsonDeserializer<>(Person.class));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Person> kafkaListenerContainerFactory() {   
        ConcurrentKafkaListenerContainerFactory<String, Person> factory =new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(receiverFactory());
        return factory;
    }
}

package fr.tutoriel.kafka.receiver;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import fr.tutoriel.kafka.dto.Person;


/**
 * Receveur Kafka
 *
 */
@Service
public class KafkaReceiver implements IReceiver {
        
        @Autowired 
        private IProcess process; // Affectation de l'instance pour la gestion des traitements
                
        /**
         * Ecoute sur le topic "Tuto1", s'il existe un message
         * 
         * @param person
         *              Objet reçu
         */
        @KafkaListener(topics = "Tuto1",containerFactory="kafkaListenerContainerFactory")
        public void listenTuto1(Person person) {
                listen("Tuto1",person); 
        }

        /**
         * Ecoute sur le topic "Tuto2", s'il existe un message

         * @param person
         *              Objet reçu
         */
        @KafkaListener(topics = "Tuto2",containerFactory="kafkaListenerContainerFactory")
        public void listenTuto2(Person person) {
                listen("Tuto2",person); 
        }
        
        /**
         *       
         * Exécution d'un traitement
         * @param topicName
         *              Nom du topic
         * @param topicName
         *          Objet reçu
         * 
         */
        @Override
        public void listen(String topicName, Person person) {
                process.execute(topicName+"Info", person);      
        }
}

package fr.tutoriel.kafka.receiver;

import org.springframework.stereotype.Service;

import fr.tutoriel.kafka.dto.Person;


/**
 * Traitement exemple
 */
@Service
public class ProcessExample implements IProcess {

        /**
         * Exécuter le traitement : Dans notre exemple affichage du message dans la console
         * @param topicName
         *              Nom du Topic
         * @param person
         *              objet person
         */
        @Override
        public void execute(String info, Person person) {
                System.out.println("Message reçu: info="+info +", person=" + person);
        }
}

package fr.tutoriel.kafka.sender;

import fr.tutoriel.kafka.dto.Person;


/**
 * Interface pour envoyer un message
 * 
 * Il serait possible d'utiliser la même interface pour envoyer 
 * un message avec une autre messagerie que Kafka (ex : RabbitMQ)
 * */
public interface ISender {
        void send(String topicName, Person person); 
}

package fr.tutoriel.kafka.sender;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
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 org.springframework.kafka.support.serializer.JsonSerializer;

import fr.tutoriel.kafka.dto.Person;

/**
 *  Configuration de l'expéditeur
 */
@Configuration
public class KafkaSenderConfig {

        @Value(value = "${kafka.bootstrapAddress:kafka:9092}") // Adresse du serveur Kafka (il est possible de le configurer via application.properties)
    private String bootstrapAddress;
    
    @Bean
    public ProducerFactory<String, Person> senderFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapAddress); // Adresse du serveur
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class);  
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,JsonSerializer.class);
        return new DefaultKafkaProducerFactory<>(configProps);
    }
    
    @Bean
    public KafkaTemplate<String, Person> kafkaTemplate() {
        return new KafkaTemplate<>(senderFactory());
    }    
}

package fr.tutoriel.kafka.sender;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.admin.AdminClientConfig;
//import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaAdmin;

/**
 *  Configuration Topic
 */
@Configuration
public class KafkaTopicConfig {
    
    @Value(value = "${kafka.bootstrapAddress:kafka:9092}") // Adresse du serveur Kafka (il est possible de le configurer via application.properties)
    private String bootstrapAddress;

    @Bean
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        return new KafkaAdmin(configs);
    }
}

package fr.tutoriel.kafka.sender;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import fr.tutoriel.kafka.dto.Person;

/**
 * 
 * Expediteur Kafka
 */
@Service
public class KafkaSender implements ISender {

        
        @Autowired
        private KafkaTemplate<String, Person> kafkaTemplate;

        /**
         * Permet d'envoyer un message (Person) dans un topic
         *   
         * @param topicName
         *                      Nom du topic
         * @param person
         *                      Objet à envoyer 
         */
        @Override
        public void send(String topicName, Person person) {
                System.out.println("Message à envoyer : topicName="+topicName +", person=" + person);           
                new NewTopic(topicName, 1, (short) 1);
                kafkaTemplate.send(topicName, person);
        }
}

package fr.tutoriel.kafka;

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

import fr.tutoriel.kafka.dto.Person;
import fr.tutoriel.kafka.sender.ISender;

/**
 * KafkaApplication 
 * Programme principal
 */
@SpringBootApplication
public class KafkaApplication implements CommandLineRunner {

        @Autowired
        private ISender sender; // Affectation de l'instance sender

        public static void main(String[] args) {
                SpringApplication.run(KafkaApplication.class, args);
        }

        /**
         * Exécuter le programme
         * Permet d'envoyer des messages vers une messagerie (Kafka ou autre)
         * 
         * La réception du message est asynchrone (en attente)
         */
        @Override
        public void run(String... args) throws Exception {
                int ageTuto1 = 10; //
                int ageTuto2 = 20; //
                while (true) { // Boucle infinie                        
                        Person personTuto1 = new Person("Jean", "DUPOND", ageTuto1); // Créer un objet pour Tuto1
                        sender.send("Tuto1", personTuto1); // Envoyer l'objet sur le topic "Tuto1"
                        Thread.sleep(3000); // Attendre 3s
                        Person personTuto2 = new Person("Pierre","DURAND",ageTuto2); // Créer un objet pour Tuto2
                        sender.send("Tuto2", personTuto2);  // Envoyer l'objet sur le topic "Tuto2"
                        Thread.sleep(3000); // Attendre 3s
                        ageTuto1++; // Incrémentation du texte
                        ageTuto2++;     // Incrémentation du texte
                }

        }
}