From: Kai Moritz Date: Mon, 28 Oct 2024 13:14:51 +0000 (+0100) Subject: `simple-producer` in `simple-consumer` verwandelt -- COPY X-Git-Tag: grundlagen/simple-consumer--2024-11-13--si~3 X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=573167bb2a80ed3b04dd28469b70046abd7d5588;p=demos%2Fkafka%2Ftraining `simple-producer` in `simple-consumer` verwandelt -- COPY --- 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..aa014a6 --- /dev/null +++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java @@ -0,0 +1,159 @@ +package de.juplo.kafka; + +import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.StringSerializer; + +import java.util.Properties; + + +@Slf4j +public class ExampleProducer +{ + private final String id; + private final String topic; + private final Producer producer; + + private volatile boolean running = true; + private volatile boolean done = false; + private long produced = 0; + + public ExampleProducer( + String broker, + String topic, + String clientId) + { + Properties props = new Properties(); + props.put("bootstrap.servers", broker); + props.put("client.id", clientId); // Nur zur Wiedererkennung + props.put("key.serializer", StringSerializer.class.getName()); + props.put("value.serializer", StringSerializer.class.getName()); + props.put("metadata.maxage.ms", 5000); + + this.id = clientId; + this.topic = topic; + producer = new KafkaProducer<>(props); + } + + public void run() + { + long i = 0; + + try + { + for (; running; i++) + { + send(Long.toString(i%10), Long.toString(i)); + Thread.sleep(500); + } + } + catch (Exception e) + { + log.error("{} - Unexpected error!", id, e); + } + finally + { + log.info("{}: Closing the KafkaProducer", id); + producer.close(); + log.info("{}: Produced {} messages in total, exiting!", id, produced); + done = true; + } + } + + void send(String key, String value) + { + final long time = System.currentTimeMillis(); + + final ProducerRecord record = new ProducerRecord<>( + topic, // Topic + key, // Key + value // Value + ); + + producer.send(record, (metadata, e) -> + { + long now = System.currentTimeMillis(); + if (e == null) + { + // HANDLE SUCCESS + produced++; + log.debug( + "{} - Sent message {}={}, partition={}:{}, timestamp={}, latency={}ms", + id, + record.key(), + record.value(), + metadata.partition(), + metadata.offset(), + metadata.timestamp(), + now - time + ); + } + else + { + // HANDLE ERROR + log.error( + "{} - ERROR for message {}={}, timestamp={}, latency={}ms: {}", + id, + record.key(), + record.value(), + metadata == null ? -1 : metadata.timestamp(), + now - time, + e.toString() + ); + } + }); + + long now = System.currentTimeMillis(); + log.trace( + "{} - Queued message {}={}, latency={}ms", + id, + record.key(), + record.value(), + now - time + ); + } + + + public static void main(String[] args) throws Exception + { + String broker = ":9092"; + String topic = "test"; + String clientId = "DEV"; + + switch (args.length) + { + case 3: + clientId = args[2]; + case 2: + topic = args[1]; + case 1: + broker = args[0]; + } + + ExampleProducer instance = new ExampleProducer(broker, topic, clientId); + + Runtime.getRuntime().addShutdownHook(new Thread(() -> + { + instance.running = false; + while (!instance.done) + { + log.info("Waiting for main-thread..."); + try + { + Thread.sleep(1000); + } + catch (InterruptedException e) {} + } + log.info("Shutdown completed."); + })); + + log.info( + "Running ExampleProducer: broker={}, topic={}, client-id={}", + broker, + topic, + clientId); + instance.run(); + } +} diff --git a/src/main/java/de/juplo/kafka/ExampleProducer.java b/src/main/java/de/juplo/kafka/ExampleProducer.java deleted file mode 100644 index aa014a6..0000000 --- a/src/main/java/de/juplo/kafka/ExampleProducer.java +++ /dev/null @@ -1,159 +0,0 @@ -package de.juplo.kafka; - -import lombok.extern.slf4j.Slf4j; -import org.apache.kafka.clients.producer.Producer; -import org.apache.kafka.clients.producer.KafkaProducer; -import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.serialization.StringSerializer; - -import java.util.Properties; - - -@Slf4j -public class ExampleProducer -{ - private final String id; - private final String topic; - private final Producer producer; - - private volatile boolean running = true; - private volatile boolean done = false; - private long produced = 0; - - public ExampleProducer( - String broker, - String topic, - String clientId) - { - Properties props = new Properties(); - props.put("bootstrap.servers", broker); - props.put("client.id", clientId); // Nur zur Wiedererkennung - props.put("key.serializer", StringSerializer.class.getName()); - props.put("value.serializer", StringSerializer.class.getName()); - props.put("metadata.maxage.ms", 5000); - - this.id = clientId; - this.topic = topic; - producer = new KafkaProducer<>(props); - } - - public void run() - { - long i = 0; - - try - { - for (; running; i++) - { - send(Long.toString(i%10), Long.toString(i)); - Thread.sleep(500); - } - } - catch (Exception e) - { - log.error("{} - Unexpected error!", id, e); - } - finally - { - log.info("{}: Closing the KafkaProducer", id); - producer.close(); - log.info("{}: Produced {} messages in total, exiting!", id, produced); - done = true; - } - } - - void send(String key, String value) - { - final long time = System.currentTimeMillis(); - - final ProducerRecord record = new ProducerRecord<>( - topic, // Topic - key, // Key - value // Value - ); - - producer.send(record, (metadata, e) -> - { - long now = System.currentTimeMillis(); - if (e == null) - { - // HANDLE SUCCESS - produced++; - log.debug( - "{} - Sent message {}={}, partition={}:{}, timestamp={}, latency={}ms", - id, - record.key(), - record.value(), - metadata.partition(), - metadata.offset(), - metadata.timestamp(), - now - time - ); - } - else - { - // HANDLE ERROR - log.error( - "{} - ERROR for message {}={}, timestamp={}, latency={}ms: {}", - id, - record.key(), - record.value(), - metadata == null ? -1 : metadata.timestamp(), - now - time, - e.toString() - ); - } - }); - - long now = System.currentTimeMillis(); - log.trace( - "{} - Queued message {}={}, latency={}ms", - id, - record.key(), - record.value(), - now - time - ); - } - - - public static void main(String[] args) throws Exception - { - String broker = ":9092"; - String topic = "test"; - String clientId = "DEV"; - - switch (args.length) - { - case 3: - clientId = args[2]; - case 2: - topic = args[1]; - case 1: - broker = args[0]; - } - - ExampleProducer instance = new ExampleProducer(broker, topic, clientId); - - Runtime.getRuntime().addShutdownHook(new Thread(() -> - { - instance.running = false; - while (!instance.done) - { - log.info("Waiting for main-thread..."); - try - { - Thread.sleep(1000); - } - catch (InterruptedException e) {} - } - log.info("Shutdown completed."); - })); - - log.info( - "Running ExampleProducer: broker={}, topic={}, client-id={}", - broker, - topic, - clientId); - instance.run(); - } -}