Spring Kafka-发出消息时如何获取时间戳(事件时间)

Question

我需要在Kafka Consumer应用程序中获得该消息时获取时间戳（事件时间）的要求。我知道可以与kafka流一起使用的时间戳extractor，但是我的要求有所不同，因为我不使用流以消耗消息。

我的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);
        }
    };

使用Spring Kafka @kafkalistener实施Kafka消费者。

@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;
    }

生产者，当我打印时，它正在发送消息，请给我以下数据：

[有效载荷= pageviewevent（userId = blog，page = air distration = 10）， 标题= {id = 8EBDAD85-E2F7-958F-500E-4560AC0970E5， kafka_messagekey = [b@71975e1a，contentType = application/json， 时间戳= 1553041963803}]

这确实具有产生的时间戳。如何使用Spring Kafka获取消息产生的时间戳记？

Answer 1

接收_timestamp意味着它是记录的时间戳，而不是收到的时间。.我们避免将其放在时间戳中，以避免无意间传播到出站消息。

Answer 2

You can use something like below:
final Producer<String, String> producer = new KafkaProducer<String, String>(properties);
        long time = System.currentTimeMillis();
        final CountDownLatch countDownLatch = new CountDownLatch(5);
        int count=0;
        try {
            for (long index = time; index < time + 10; index++) {
                String key = null;
                count++;
                if(count<=5)
                    key = "id_"+ Integer.toString(1);
                else
                    key = "id_"+ Integer.toString(2);
                final ProducerRecord<String, String> record =
                        new ProducerRecord<>(TOPIC, key, "B2B Sample Message: " + count);
                producer.send(record, (metadata, exception) -> {
                    long elapsedTime = System.currentTimeMillis() - time;
                    if (metadata != null) {
                        System.out.printf("sent record(key=%s value=%s) " +
                                        "meta(partition=%d, offset=%d) time=%d timestamp=%dn",
                                record.key(), record.value(), metadata.partition(),
                                metadata.offset(), elapsedTime, metadata.timestamp());
                        System.out.println("Timestamp:: "+metadata.timestamp() );
                    } else {
                        exception.printStackTrace();
                    }
                    countDownLatch.countDown();
                });
            }
            try {
                countDownLatch.await(25, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }finally {
            producer.flush();
            producer.close();
        }
    }