[ISSUE apache#3288]Refactor ConsumerGroupManager

mxsm · Feb 27, 2023 · be99e0f · be99e0f
1 parent 8a014d4
commit be99e0f
Showing 1 changed file with 20 additions and 13 deletions.
diff --git a/...n/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/ConsumerGroupManager.java b/...n/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/ConsumerGroupManager.java
@@ -29,31 +29,36 @@
 
 public class ConsumerGroupManager {
 
-    protected AtomicBoolean started = new AtomicBoolean(Boolean.FALSE);
+    private final AtomicBoolean started = new AtomicBoolean(false);
 
-    protected AtomicBoolean inited = new AtomicBoolean(Boolean.FALSE);
+    private final AtomicBoolean inited = new AtomicBoolean(false);
 
-    private EventMeshHTTPServer eventMeshHTTPServer;
+    private final EventMeshHTTPServer eventMeshHTTPServer;
 
-    private EventMeshConsumer eventMeshConsumer;
+    private final EventMeshConsumer eventMeshConsumer;
 
     private ConsumerGroupConf consumerGroupConfig;
 
-    public ConsumerGroupManager(EventMeshHTTPServer eventMeshHTTPServer, ConsumerGroupConf consumerGroupConfig) {
+    public ConsumerGroupManager(final EventMeshHTTPServer eventMeshHTTPServer, final ConsumerGroupConf consumerGroupConfig) {
         this.eventMeshHTTPServer = eventMeshHTTPServer;
         this.consumerGroupConfig = consumerGroupConfig;
-        eventMeshConsumer = new EventMeshConsumer(this.eventMeshHTTPServer, this.consumerGroupConfig);
+        this.eventMeshConsumer = new EventMeshConsumer(this.eventMeshHTTPServer, this.consumerGroupConfig);
     }
 
-    public synchronized void init() throws Exception {
+    public void init() throws Exception {
+        if (!inited.compareAndSet(false, true)) {
+            return;
+        }
         eventMeshConsumer.init();
-        inited.compareAndSet(false, true);
+
     }
 
-    public synchronized void start() throws Exception {
+    public void start() throws Exception {
+        if (!started.compareAndSet(false, true)) {
+            return;
+        }
         setupEventMeshConsumer(consumerGroupConfig);
         eventMeshConsumer.start();
-        started.compareAndSet(false, true);
     }
 
     private synchronized void setupEventMeshConsumer(ConsumerGroupConf consumerGroupConfig) throws Exception {
@@ -62,12 +67,14 @@ private synchronized void setupEventMeshConsumer(ConsumerGroupConf consumerGroup
         }
     }
 
-    public synchronized void shutdown() throws Exception {
+    public void shutdown() throws Exception {
+        if (!started.compareAndSet(true, false)) {
+            return;
+        }
         eventMeshConsumer.shutdown();
-        started.compareAndSet(true, false);
     }
 
-    public synchronized void refresh(ConsumerGroupConf consumerGroupConfig) throws Exception {
+    public synchronized void refresh(final ConsumerGroupConf consumerGroupConfig) throws Exception {
 
         if (consumerGroupConfig == null || this.consumerGroupConfig.equals(consumerGroupConfig)) {
             return;