From: Kai Moritz Date: Wed, 25 Sep 2024 11:57:49 +0000 (+0200) Subject: Einheitliche Benennung des Consumers -- MOVE X-Git-Tag: grundlagen/simple-consumer--BRANCH-ENDE~9 X-Git-Url: https://juplo.de/gitweb/?a=commitdiff_plain;h=f0fa6ee6439f8a283b69d087f3ebc5ecf8d3d9be;p=demos%2Fkafka%2Ftraining Einheitliche Benennung des Consumers -- MOVE --- diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java new file mode 100644 index 0000000..cee2165 --- /dev/null +++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java @@ -0,0 +1,137 @@ +package de.juplo.kafka; + +import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.common.errors.WakeupException; +import org.apache.kafka.common.serialization.StringDeserializer; + +import java.time.Duration; +import java.util.Arrays; +import java.util.Properties; + + +@Slf4j +public class SimpleConsumer +{ + private final String id; + private final String topic; + private final Consumer consumer; + + private volatile boolean running = false; + private long consumed = 0; + + public SimpleConsumer(String broker, String topic, String groupId, String clientId) + { + Properties props = new Properties(); + props.put("bootstrap.servers", broker); + props.put("group.id", groupId); // ID für die Offset-Commits + props.put("client.id", clientId); // Nur zur Wiedererkennung + props.put("auto.offset.reset", "earliest"); // Von Beginn an lesen + props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.CooperativeStickyAssignor"); + props.put("key.deserializer", StringDeserializer.class.getName()); + props.put("value.deserializer", StringDeserializer.class.getName()); + + this.id = clientId; + this.topic = topic; + consumer = new KafkaConsumer<>(props); + } + + + public void run() + { + try + { + log.info("{} - Subscribing to topic {}", id, topic); + consumer.subscribe(Arrays.asList(topic)); + running = true; + + while (true) + { + ConsumerRecords records = + consumer.poll(Duration.ofSeconds(1)); + + 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() + ); + } + } + } + catch(WakeupException e) + { + log.info("{} - Consumer was signaled to finish its work", id); + } + catch(Exception e) + { + log.error("{} - Unexpected error: {}, unsubscribing!", id, e.toString()); + consumer.unsubscribe(); + } + finally + { + running = false; + log.info("{} - Closing the KafkaConsumer", id); + consumer.close(); + log.info("{}: Consumed {} messages in total, exiting!", id, consumed); + } + } + + + public static void main(String[] args) throws Exception + { + String broker = ":9092"; + String topic = "test"; + String groupId = "my-group"; + String clientId = "DEV"; + + switch (args.length) + { + case 4: + clientId = args[3]; + case 3: + groupId = args[2]; + case 2: + topic = args[1]; + case 1: + broker = args[0]; + } + + + SimpleConsumer instance = new SimpleConsumer(broker, topic, groupId, clientId); + + Runtime.getRuntime().addShutdownHook(new Thread(() -> + { + instance.consumer.wakeup(); + + while (instance.running) + { + log.info("Waiting for main-thread..."); + try + { + Thread.sleep(1000); + } + catch (InterruptedException e) {} + } + log.info("Shutdown completed."); + })); + + log.info( + "Running SimpleConsumer: broker={}, topic={}, group-id={}, client-id={}", + broker, + topic, + groupId, + clientId); + instance.run(); + } +} diff --git a/src/main/java/de/juplo/kafka/SimpleConsumer.java b/src/main/java/de/juplo/kafka/SimpleConsumer.java deleted file mode 100644 index cee2165..0000000 --- a/src/main/java/de/juplo/kafka/SimpleConsumer.java +++ /dev/null @@ -1,137 +0,0 @@ -package de.juplo.kafka; - -import lombok.extern.slf4j.Slf4j; -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.ConsumerRecords; -import org.apache.kafka.clients.consumer.Consumer; -import org.apache.kafka.clients.consumer.KafkaConsumer; -import org.apache.kafka.common.errors.WakeupException; -import org.apache.kafka.common.serialization.StringDeserializer; - -import java.time.Duration; -import java.util.Arrays; -import java.util.Properties; - - -@Slf4j -public class SimpleConsumer -{ - private final String id; - private final String topic; - private final Consumer consumer; - - private volatile boolean running = false; - private long consumed = 0; - - public SimpleConsumer(String broker, String topic, String groupId, String clientId) - { - Properties props = new Properties(); - props.put("bootstrap.servers", broker); - props.put("group.id", groupId); // ID für die Offset-Commits - props.put("client.id", clientId); // Nur zur Wiedererkennung - props.put("auto.offset.reset", "earliest"); // Von Beginn an lesen - props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.CooperativeStickyAssignor"); - props.put("key.deserializer", StringDeserializer.class.getName()); - props.put("value.deserializer", StringDeserializer.class.getName()); - - this.id = clientId; - this.topic = topic; - consumer = new KafkaConsumer<>(props); - } - - - public void run() - { - try - { - log.info("{} - Subscribing to topic {}", id, topic); - consumer.subscribe(Arrays.asList(topic)); - running = true; - - while (true) - { - ConsumerRecords records = - consumer.poll(Duration.ofSeconds(1)); - - 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() - ); - } - } - } - catch(WakeupException e) - { - log.info("{} - Consumer was signaled to finish its work", id); - } - catch(Exception e) - { - log.error("{} - Unexpected error: {}, unsubscribing!", id, e.toString()); - consumer.unsubscribe(); - } - finally - { - running = false; - log.info("{} - Closing the KafkaConsumer", id); - consumer.close(); - log.info("{}: Consumed {} messages in total, exiting!", id, consumed); - } - } - - - public static void main(String[] args) throws Exception - { - String broker = ":9092"; - String topic = "test"; - String groupId = "my-group"; - String clientId = "DEV"; - - switch (args.length) - { - case 4: - clientId = args[3]; - case 3: - groupId = args[2]; - case 2: - topic = args[1]; - case 1: - broker = args[0]; - } - - - SimpleConsumer instance = new SimpleConsumer(broker, topic, groupId, clientId); - - Runtime.getRuntime().addShutdownHook(new Thread(() -> - { - instance.consumer.wakeup(); - - while (instance.running) - { - log.info("Waiting for main-thread..."); - try - { - Thread.sleep(1000); - } - catch (InterruptedException e) {} - } - log.info("Shutdown completed."); - })); - - log.info( - "Running SimpleConsumer: broker={}, topic={}, group-id={}, client-id={}", - broker, - topic, - groupId, - clientId); - instance.run(); - } -}