1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
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)+"}";        
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
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);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
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);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
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;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
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);	
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
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);
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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); 
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
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());
    }    
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
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);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
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);
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
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
		}

	}
}

1
2
3
4
5
6
7
Message à envoyer : topicName=Tuto2, person={firstName:"Pierre", lastName:"DURAND", age:20}
Message reçu: info=Tuto2Info, person={firstName:"Pierre", lastName:"DURAND", age:20}
Message à envoyer : topicName=Tuto1, person={firstName:"Jean", lastName:"DUPOND", age:11}
Message reçu: info=Tuto1Info, person={firstName:"Jean", lastName:"DUPOND", age:11}
Message à envoyer : topicName=Tuto2, person={firstName:"Pierre", lastName:"DURAND", age:21}
Message reçu: info=Tuto2Info, person={firstName:"Pierre", lastName:"DURAND", age:21}