Получение ключа с использованием SpringBoot Kafka Consumer с использованием класса ConsumerFactory - PullRequest
Новая
       15

Получение ключа с использованием SpringBoot Kafka Consumer с использованием класса ConsumerFactory

0 голосов
/ 28 января 2019 
 @Bean
    public ConsumerFactory<String,String> consumerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configProps.put("auto.offset.reset", "earliest");
        return new DefaultKafkaConsumerFactory<>(configProps);
    }

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

    @Bean
    public Receiver receiver(){
        return  new Receiver();
    }


**Receiver class :**
    public class Receiver {

    @KafkaListener(topics = "topicname")
    public void listen(String message) {System.out.println("Kakfa Listener received message " + message);

    }
}

Проблема в том, что аргумент "String message" выдает мне только сообщение, а не ключ, соответствующий сообщению Kafka.Я хотел бы получить как ключ, так и сообщение.Пожалуйста, ответьте, если есть способ получить значение ключа.

...