juplo.de Git - demos/kafka/training/blob - src/main/java/de/juplo/kafka/SimpleConsumer.java

   1 package de.juplo.kafka;
   2
   3 import lombok.extern.slf4j.Slf4j;
   4 import org.apache.kafka.clients.consumer.ConsumerRecord;
   5 import org.apache.kafka.clients.consumer.ConsumerRecords;
   6 import org.apache.kafka.clients.consumer.KafkaConsumer;
   7 import org.apache.kafka.common.serialization.StringDeserializer;
   8
   9 import java.time.Duration;
  10 import java.util.Arrays;
  11 import java.util.Properties;
  12
  13
  14 @Slf4j
  15 public class SimpleConsumer
  16 {
  17   private final String id;
  18   private final String topic;
  19   private final KafkaConsumer<String, String> consumer;
  20
  21   private long consumed = 0;
  22   private volatile boolean running = true;
  23   private volatile boolean done = false;
  24
  25   public SimpleConsumer(String broker, String topic, String groupId, String clientId)
  26   {
  27     Properties props = new Properties();
  28     props.put("bootstrap.servers", broker);
  29     props.put("group.id", groupId); // ID für die Offset-Commits
  30     props.put("client.id", clientId); // Nur zur Wiedererkennung
  31     props.put("key.deserializer", StringDeserializer.class.getName());
  32     props.put("value.deserializer", StringDeserializer.class.getName());
  33
  34     consumer = new KafkaConsumer<>(props);
  35
  36     this.topic = topic;
  37     this.id = clientId;
  38   }
  39
  40
  41   public void run()
  42   {
  43     try
  44     {
  45       log.info("{} - Subscribing to topic test", id);
  46       consumer.subscribe(Arrays.asList("test"));
  47
  48       // tag::loop[]
  49       while (running)
  50       {
  51         ConsumerRecords<String, String> records =
  52             consumer.poll(Duration.ofSeconds(1));
  53
  54         // Do something with the data...
  55         // end::loop[]
  56         log.info("{} - Received {} messages", id, records.count());
  57         for (ConsumerRecord<String, String> record : records)
  58         {
  59           consumed++;
  60           log.info(
  61               "{} - {}: {}/{} - {}={}",
  62               id,
  63               record.offset(),
  64               record.topic(),
  65               record.partition(),
  66               record.key(),
  67               record.value()
  68           );
  69         }
  70         // tag::loop[]
  71       }
  72       // end::loop[]
  73     }
  74     catch(Exception e)
  75     {
  76       log.error("{} - Unexpected error: {}", id, e.toString());
  77     }
  78     finally
  79     {
  80       log.info("{} - Closing the KafkaConsumer", id);
  81       consumer.close();
  82       log.info("{}: Consumed {} messages in total, exiting!", id, consumed);
  83       done = true;
  84     }
  85   }
  86
  87
  88   public static void main(String[] args) throws Exception
  89   {
  90     String broker = ":9092";
  91     String topic = "test";
  92     String groupId = "my-group";
  93     String clientId = "DEV";
  94
  95     switch (args.length)
  96     {
  97       case 4:
  98         clientId = args[3];
  99       case 3:
 100         groupId = args[2];
 101       case 2:
 102         topic = args[1];
 103       case 1:
 104         broker = args[0];
 105     }
 106
 107
 108     SimpleConsumer instance = new SimpleConsumer(broker, topic, groupId, clientId);
 109
 110     Runtime.getRuntime().addShutdownHook(new Thread(() ->
 111     {
 112       instance.running = false;
 113       while (!instance.done)
 114       {
 115         log.info("Waiting for main-thread...");
 116         try
 117         {
 118           Thread.sleep(1000);
 119         }
 120         catch (InterruptedException e) {}
 121       }
 122       log.info("Shutdown completed.");
 123     }));
 124
 125     log.info(
 126         "Running SimpleConsumer: broker={}, topic={}, group-id={}, client-id={}",
 127         broker,
 128         topic,
 129         groupId,
 130         clientId);
 131     instance.run();
 132   }
 133 }