fix multi thread close bug

leonchen83 · leonchen83 · commit 578bc4905129 · 2017-11-07T18:04:05.000+08:00
diff --git a/rocketmq-redis/src/main/java/org/apache/rocketmq/redis/replicator/AbstractReplicator.java b/rocketmq-redis/src/main/java/org/apache/rocketmq/redis/replicator/AbstractReplicator.java
@@ -275,8 +275,13 @@ public void builtInCommandParserRegister() {
         addCommandParser(CommandName.name("RPOPLPUSH"), new RPopLPushParser());
     }
 
+    @Override
+    public void close() throws IOException {
+        this.connected.compareAndSet(CONNECTED, DISCONNECTING);
+    }
+
     protected void doClose() throws IOException {
-        if (!this.connected.compareAndSet(CONNECTED, DISCONNECTING)) return;
+        this.connected.compareAndSet(CONNECTED, DISCONNECTING);
         try {
             if (inputStream != null) {
                 this.inputStream.setRawByteListeners(null);
@@ -287,6 +292,5 @@ protected void doClose() throws IOException {
         } finally {
             this.connected.set(DISCONNECTED);
         }
-        doCloseListener(this);
     }
 }
diff --git a/rocketmq-redis/src/main/java/org/apache/rocketmq/redis/replicator/RedisAofReplicator.java b/rocketmq-redis/src/main/java/org/apache/rocketmq/redis/replicator/RedisAofReplicator.java
@@ -68,7 +68,8 @@ public void open() throws IOException {
             if (!(e.getCause() instanceof EOFException))
                 throw e.getCause();
         } finally {
-            close();
+            doClose();
+            doCloseListener(this);
         }
     }
 
@@ -97,10 +98,4 @@ protected void doOpen() throws IOException {
             }
         }
     }
-
-    @Override
-    public void close() throws IOException {
-        doClose();
-    }
-
 }
diff --git a/rocketmq-redis/src/main/java/org/apache/rocketmq/redis/replicator/RedisMixReplicator.java b/rocketmq-redis/src/main/java/org/apache/rocketmq/redis/replicator/RedisMixReplicator.java
@@ -75,7 +75,8 @@ public void open() throws IOException {
             if (!(e.getCause() instanceof EOFException))
                 throw e.getCause();
         } finally {
-            close();
+            doClose();
+            doCloseListener(this);
         }
     }
 
@@ -107,9 +108,4 @@ protected void doOpen() throws IOException {
             }
         }
     }
-
-    @Override
-    public void close() throws IOException {
-        doClose();
-    }
 }
diff --git a/rocketmq-redis/src/main/java/org/apache/rocketmq/redis/replicator/RedisRdbReplicator.java b/rocketmq-redis/src/main/java/org/apache/rocketmq/redis/replicator/RedisRdbReplicator.java
@@ -56,18 +56,13 @@ public void open() throws IOException {
             if (!(e.getCause() instanceof EOFException))
                 throw e.getCause();
         } finally {
-            close();
+            doClose();
+            doCloseListener(this);
         }
     }
 
     protected void doOpen() throws IOException {
         RdbParser parser = new RdbParser(inputStream, this);
         parser.parse();
     }
-
-    @Override
-    public void close() throws IOException {
-        doClose();
-    }
-
 }
diff --git a/rocketmq-redis/src/main/java/org/apache/rocketmq/redis/replicator/RedisSocketReplicator.java b/rocketmq-redis/src/main/java/org/apache/rocketmq/redis/replicator/RedisSocketReplicator.java
@@ -83,7 +83,7 @@ public void open() throws IOException {
         try {
             doOpen();
         } finally {
-            close();
+            doClose();
             doCloseListener(this);
         }
     }
@@ -112,7 +112,7 @@ protected void doOpen() throws IOException {
                     heartbeat();
                 } else if (syncMode == SyncMode.SYNC_LATER && getStatus() == CONNECTED) {
                     i = 0;
-                    close();
+                    doClose();
                     try {
                         Thread.sleep(configuration.getRetryTimeInterval());
                     } catch (InterruptedException interrupt) {
@@ -167,7 +167,7 @@ public void handle(long len) {
                     exception = (IOException) e;
                 }
                 LOGGER.error("[redis-replicator] socket error", exception);
-                close();
+                doClose();
                 //retry psync in next loop.
                 if (LOGGER.isInfoEnabled()) {
                     LOGGER.info("reconnect to redis-server. retry times:" + (i + 1));
@@ -404,9 +404,8 @@ protected void connect() throws IOException {
     }
 
     @Override
-    public void close() {
-        if (!connected.compareAndSet(CONNECTED, DISCONNECTING))
-            return;
+    protected void doClose() throws IOException {
+        connected.compareAndSet(CONNECTED, DISCONNECTING);
 
         try {
             synchronized (this) {

Original file line number	Diff line number	Diff line change
`@@ -275,8 +275,13 @@ public void builtInCommandParserRegister() {`
`275`	`275`	`addCommandParser(CommandName.name("RPOPLPUSH"), new RPopLPushParser());`
`276`	`276`	`}`
`277`	`277`
	`278`	`+ @Override`
	`279`	`+ public void close() throws IOException {`
	`280`	`+ this.connected.compareAndSet(CONNECTED, DISCONNECTING);`
	`281`	`+ }`
	`282`	`+`
`278`	`283`	`protected void doClose() throws IOException {`
`279`		`- if (!this.connected.compareAndSet(CONNECTED, DISCONNECTING)) return;`
	`284`	`+ this.connected.compareAndSet(CONNECTED, DISCONNECTING);`
`280`	`285`	`try {`
`281`	`286`	`if (inputStream != null) {`
`282`	`287`	`this.inputStream.setRawByteListeners(null);`
`@@ -287,6 +292,5 @@ protected void doClose() throws IOException {`
`287`	`292`	`} finally {`
`288`	`293`	`this.connected.set(DISCONNECTED);`
`289`	`294`	`}`
`290`		`- doCloseListener(this);`
`291`	`295`	`}`
`292`	`296`	`}`
Original file line number	Diff line number	Diff line change
`@@ -68,7 +68,8 @@ public void open() throws IOException {`
`68`	`68`	`if (!(e.getCause() instanceof EOFException))`
`69`	`69`	`throw e.getCause();`
`70`	`70`	`} finally {`
`71`		`- close();`
	`71`	`+ doClose();`
	`72`	`+ doCloseListener(this);`
`72`	`73`	`}`
`73`	`74`	`}`
`74`	`75`
`@@ -97,10 +98,4 @@ protected void doOpen() throws IOException {`
`97`	`98`	`}`
`98`	`99`	`}`
`99`	`100`	`}`
`100`		`-`
`101`		`- @Override`
`102`		`- public void close() throws IOException {`
`103`		`- doClose();`
`104`		`- }`
`105`		`-`
`106`	`101`	`}`
Original file line number	Diff line number	Diff line change
`@@ -75,7 +75,8 @@ public void open() throws IOException {`
`75`	`75`	`if (!(e.getCause() instanceof EOFException))`
`76`	`76`	`throw e.getCause();`
`77`	`77`	`} finally {`
`78`		`- close();`
	`78`	`+ doClose();`
	`79`	`+ doCloseListener(this);`
`79`	`80`	`}`
`80`	`81`	`}`
`81`	`82`
`@@ -107,9 +108,4 @@ protected void doOpen() throws IOException {`
`107`	`108`	`}`
`108`	`109`	`}`
`109`	`110`	`}`
`110`		`-`
`111`		`- @Override`
`112`		`- public void close() throws IOException {`
`113`		`- doClose();`
`114`		`- }`
`115`	`111`	`}`
Original file line number	Diff line number	Diff line change
`@@ -56,18 +56,13 @@ public void open() throws IOException {`
`56`	`56`	`if (!(e.getCause() instanceof EOFException))`
`57`	`57`	`throw e.getCause();`
`58`	`58`	`} finally {`
`59`		`- close();`
	`59`	`+ doClose();`
	`60`	`+ doCloseListener(this);`
`60`	`61`	`}`
`61`	`62`	`}`
`62`	`63`
`63`	`64`	`protected void doOpen() throws IOException {`
`64`	`65`	`RdbParser parser = new RdbParser(inputStream, this);`
`65`	`66`	`parser.parse();`
`66`	`67`	`}`
`67`		`-`
`68`		`- @Override`
`69`		`- public void close() throws IOException {`
`70`		`- doClose();`
`71`		`- }`
`72`		`-`
`73`	`68`	`}`