From: Kai Moritz Date: Sat, 12 Nov 2022 14:01:00 +0000 (+0100) Subject: Konsumption der Nachricht in eine separate Methode verlagert X-Git-Tag: spring-consumer--topicpartition-DEPRECATED~5 X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=8f14291a52872141ded7ecead6b1f7f9fbc969e4;p=demos%2Fkafka%2Ftraining Konsumption der Nachricht in eine separate Methode verlagert * Dies ist eine Vorbereitung für die Umstellung auf `@KafkaListener` * Durch die Umstellung wird der Code vor/nach der Umstellung besser vergleichbar. * Denn mit `@KafkaListener` wird direkt eine Methode annotiert, die dann von Spring die deserialisierte Nachricht injiziert bekommt. --- diff --git a/src/main/java/de/juplo/kafka/SimpleConsumer.java b/src/main/java/de/juplo/kafka/SimpleConsumer.java index 0e37686..1cf9b22 100644 --- a/src/main/java/de/juplo/kafka/SimpleConsumer.java +++ b/src/main/java/de/juplo/kafka/SimpleConsumer.java @@ -38,16 +38,12 @@ public class SimpleConsumer implements Runnable log.info("{} - Received {} messages", id, records.count()); for (ConsumerRecord record : records) { - consumed++; - log.info( - "{} - {}: {}/{} - {}={}", - id, - record.offset(), - record.topic(), - record.partition(), - record.key(), - record.value() - ); + handleRecord( + record.topic(), + record.partition(), + record.offset(), + record.key(), + record.value()); } } } @@ -67,4 +63,15 @@ public class SimpleConsumer implements Runnable log.info("{}: Consumed {} messages in total, exiting!", id, consumed); } } + + private void handleRecord( + String topic, + Integer partition, + Long offset, + String key, + String value) + { + consumed++; + log.info("{} - {}: {}/{} - {}={}", id, offset, topic, partition, key, value); + } }