Some improvement made for AsyncSemaphore object

redisson/src/main/java/org/redisson/connection/ClientConnectionsEntry.java

@@ -113,6 +113,10 @@ public class ClientConnectionsEntry {
     public void acquireConnection(Runnable runnable) {
         freeConnectionsCounter.acquire(runnable);
     }
+    
+    public void removeConnection(Runnable runnable) {
+        freeConnectionsCounter.remove(runnable);
+    }
 
     public void releaseConnection() {
         freeConnectionsCounter.release();

redisson/src/main/java/org/redisson/connection/pool/ConnectionPool.java

@@ -20,6 +20,7 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.redisson.api.NodeType;
@@ -201,14 +202,29 @@ abstract class ConnectionPool<T extends RedisConnection> {
         return connectionManager.newFailedFuture(exception);
     }
 
+    public static abstract class AcquireCallback<T> implements Runnable, FutureListener<T> {
+        
+    }
+    
     private RFuture<T> acquireConnection(RedisCommand<?> command, final ClientConnectionsEntry entry) {
         final RPromise<T> result = connectionManager.newPromise();
-        acquireConnection(entry, new Runnable() {
+
+        AcquireCallback<T> callback = new AcquireCallback<T>() {
             @Override
             public void run() {
+                result.removeListener(this);
                 connectTo(entry, result);
             }
-        });
+            
+            @Override
+            public void operationComplete(Future<T> future) throws Exception {
+                entry.removeConnection(this);
+            }
+        };
+        
+        result.addListener(callback);
+        acquireConnection(entry, callback);
+        
         return result;
     }
 

redisson/src/main/java/org/redisson/pubsub/AsyncSemaphore.java

@@ -15,8 +15,11 @@
  */
 package org.redisson.pubsub;
 
+import java.util.Iterator;
+import java.util.LinkedHashSet;
 import java.util.LinkedList;
 import java.util.Queue;
+import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 
 /**
@@ -27,7 +30,7 @@ import java.util.concurrent.CountDownLatch;
 public class AsyncSemaphore {
 
     private int counter;
-    private final Queue<Runnable> listeners = new LinkedList<Runnable>();
+    private final Set<Runnable> listeners = new LinkedHashSet<Runnable>();
 
     public AsyncSemaphore(int permits) {
         counter = permits;
@@ -89,7 +92,11 @@ public class AsyncSemaphore {
         
         synchronized (this) {
             counter++;
-            runnable = listeners.poll();
+            Iterator<Runnable> iter = listeners.iterator();
+            if (iter.hasNext()) {
+                runnable = iter.next();
+                iter.remove();
+            }
         }
         
         if (runnable != null) {