From: Kai Moritz Date: Sun, 10 Apr 2022 16:58:16 +0000 (+0200) Subject: Thread-Synchronisation bei start/stop überarbeitet X-Git-Tag: deserialization-synchroner-test~17^2 X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=014de600554a5f9ba620fdadb8581b5882bc4d20;p=demos%2Fkafka%2Ftraining Thread-Synchronisation bei start/stop überarbeitet --- diff --git a/src/main/java/de/juplo/kafka/EndlessConsumer.java b/src/main/java/de/juplo/kafka/EndlessConsumer.java index f25e93c..18cef23 100644 --- a/src/main/java/de/juplo/kafka/EndlessConsumer.java +++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java @@ -13,8 +13,9 @@ import java.util.Arrays; import java.util.Properties; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; -import java.util.concurrent.Future; -import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; @Slf4j @@ -27,10 +28,12 @@ public class EndlessConsumer implements Runnable private final String topic; private final String autoOffsetReset; - private AtomicBoolean running = new AtomicBoolean(); + private final Lock lock = new ReentrantLock(); + private final Condition condition = lock.newCondition(); + private boolean running = false; private long consumed = 0; private KafkaConsumer consumer = null; - private Future future = null; + public EndlessConsumer( ExecutorService executor, @@ -91,11 +94,12 @@ public class EndlessConsumer implements Runnable catch(WakeupException e) { log.info("{} - RIIING!", id); + shutdown(); } catch(Exception e) { log.error("{} - Unexpected error: {}", id, e.toString(), e); - running.set(false); // Mark the instance as not running + shutdown(); } finally { @@ -105,27 +109,55 @@ public class EndlessConsumer implements Runnable } } + private void shutdown() + { + lock.lock(); + try + { + running = false; + condition.signal(); + } + finally + { + lock.unlock(); + } + } - public synchronized void start() + public void start() { - boolean stateChanged = running.compareAndSet(false, true); - if (!stateChanged) - throw new IllegalStateException("Consumer instance " + id + " is already running!"); + lock.lock(); + try + { + if (running) + throw new IllegalStateException("Consumer instance " + id + " is already running!"); - log.info("{} - Starting - consumed {} messages before", id, consumed); - future = executor.submit(this); + log.info("{} - Starting - consumed {} messages before", id, consumed); + running = true; + executor.submit(this); + } + finally + { + lock.unlock(); + } } public synchronized void stop() throws ExecutionException, InterruptedException { - boolean stateChanged = running.compareAndSet(true, false); - if (!stateChanged) - throw new IllegalStateException("Consumer instance " + id + " is not running!"); - - log.info("{} - Stopping", id); - consumer.wakeup(); - future.get(); - log.info("{} - Stopped - consumed {} messages so far", id, consumed); + lock.lock(); + try + { + if (!running) + throw new IllegalStateException("Consumer instance " + id + " is not running!"); + + log.info("{} - Stopping", id); + consumer.wakeup(); + condition.await(); + log.info("{} - Stopped - consumed {} messages so far", id, consumed); + } + finally + { + lock.unlock(); + } } @PreDestroy