Spring Kafka - как получить метку времени (время события), когда было создано сообщение - PullRequest
Новая
       34

Spring Kafka - как получить метку времени (время события), когда было создано сообщение

1 голос
/ 20 марта 2019

У меня есть требование получить метку времени (событие-время), когда сообщение было сгенерировано, в приложении-потребителе kafka.Мне известно о timestampExtractor, который можно использовать с потоком kafka, но у меня другое требование, поскольку я не использую поток для получения сообщения.

Мой производитель kafka выглядит следующим образом: 

@Override
public void run(ApplicationArguments args) throws Exception {


    List<String> names = Arrays.asList("priya", "dyser", "Ray", "Mark", "Oman", "Larry");
    List<String> pages = Arrays.asList("blog", "facebook", "instagram", "news", "youtube", "about");
    Runnable runnable = () -> {
        String rPage = pages.get(new Random().nextInt(pages.size()));
        String rName = pages.get(new Random().nextInt(names.size()));
        PageViewEvent pageViewEvent = new PageViewEvent(rName, rPage, Math.random() > .5 ? 10 : 1000);

        Message<PageViewEvent> message =  MessageBuilder
                .withPayload(pageViewEvent).
                setHeader(KafkaHeaders.MESSAGE_KEY, pageViewEvent.getUserId().getBytes())
                        .build();

        try {
            this.pageViewsOut.send(message);
            log.info("sent " + message);
        } catch (Exception e) {
            log.error(e);
        }
    };

Kafka Consumer реализован с использованием Spring kafka @ KafkaListener. 

@KafkaListener(topics = "test1" , groupId = "json", containerFactory = "kafkaListenerContainerFactory")

    public void receive(@Payload PageViewEvent data,@Headers MessageHeaders headers) {
        LOG.info("Message received");
        LOG.info("received data='{}'", data);
 }

Заводская конфигурация контейнера 

   @Bean
   public ConsumerFactory<String,PageViewEvent > priceEventConsumerFactory() {

        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "json");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), new JsonDeserializer<>(PageViewEvent.class));



    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, PageViewEvent> priceEventsKafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, PageViewEvent> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(priceEventConsumerFactory());
        return factory;
    }

Производитель, который отправляет сообщение при печати, дает мне следующие данные:

[payload = PageViewEvent (userId = blog, page = about, duration = 10), заголовки = {id = 8ebdad85-e2f7-958f-500e-4560ac0970e5, kafka_messageKey = [B @ 71975e1a, contentType =application / json, timestamp = 1553041963803}]

У него есть созданная временная метка.Как я могу получить метку времени, созданную в сообщении, с помощью Spring kafka?

1 Ответ

1 голос
/ 20 марта 2019

RECEIVED_TIMESTAMP означает, что это отметка времени из записи, которая была получена, а не время, когда она была получена. Мы избегаем помещения ее в TIMESTAMP, чтобы избежать непреднамеренного распространения на исходящее сообщение.

...