`@KafkaListener` mit paraleller Verarbeitung

diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java
index 5894a10..58a249e 100644 (file)
--- a/src/main/java/de/juplo/kafka/ExampleConsumer.java
+++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java
@@ -18,7 +18,7 @@ public class ExampleConsumer
   private String id;
   private long consumed = 0;
 
-  @KafkaListener(topics = "${juplo.consumer.topic}")
+  @KafkaListener(topics = "${juplo.consumer.topic}", concurrency = "2")
   private void receive(
     @Header(KafkaHeaders.RECEIVED_TOPIC)
     String topic,
@@ -41,6 +41,9 @@ public class ExampleConsumer
     String key,
     String value)
   {
+    // BEACHTE:
+    // Der Zugriff auf die Variable "consumed" ist hier so nicht mehr korrekt!
+    // Grund: Der Zugriff erfolgt ohne Locking konkurrierend aus zwei Threads.
     consumed++;
     log.info("{} - partition={}-{}, offset={}: {}={}", id, topic, partition, offset, key, value);
   }