46 lines
1.9 KiB
Java
46 lines
1.9 KiB
Java
package com.baeldung.spring.kafka;
|
|
|
|
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;
|
|
|
|
@EnableKafka
|
|
@Configuration
|
|
public class KafkaConsumerConfig {
|
|
|
|
@Value(value = "${kafka.bootstrapAddress}")
|
|
private String bootstrapAddress;
|
|
|
|
public ConsumerFactory<String, String> consumerFactory(String groupId) {
|
|
Map<String, Object> props = new HashMap<>();
|
|
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
|
|
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
|
|
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
|
|
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
|
|
return new DefaultKafkaConsumerFactory<>(props);
|
|
}
|
|
|
|
@Bean
|
|
public ConcurrentKafkaListenerContainerFactory<String, String> fooKafkaListenerContainerFactory() {
|
|
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
|
|
factory.setConsumerFactory(consumerFactory("foo"));
|
|
return factory;
|
|
}
|
|
|
|
@Bean
|
|
public ConcurrentKafkaListenerContainerFactory<String, String> barKafkaListenerContainerFactory() {
|
|
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
|
|
factory.setConsumerFactory(consumerFactory("bar"));
|
|
return factory;
|
|
}
|
|
}
|