diff --git a/src/main/java/org/redisson/connection/pool/ConnectionPool.java b/src/main/java/org/redisson/connection/pool/ConnectionPool.java index 5cd79c490..d8d2a3f9c 100644 --- a/src/main/java/org/redisson/connection/pool/ConnectionPool.java +++ b/src/main/java/org/redisson/connection/pool/ConnectionPool.java @@ -79,38 +79,54 @@ abstract class ConnectionPool { } final AtomicInteger initializedConnections = new AtomicInteger(minimumIdleSize); - for (int i = 0; i < minimumIdleSize; i++) { - if ((checkFreezed && entry.isFreezed()) || !tryAcquireConnection(entry)) { - Throwable cause = new RedisConnectionException( - "Can't init enough connections amount! Only " + (minimumIdleSize - initializedConnections.get()) + " from " + minimumIdleSize + " were initialized. Server: " - + entry.getClient().getAddr()); - initPromise.tryFailure(cause); - return; - } + int startAmount = Math.min(50, minimumIdleSize); + final AtomicInteger requests = new AtomicInteger(startAmount); + for (int i = 0; i < startAmount; i++) { + createConnection(checkFreezed, requests, entry, initPromise, minimumIdleSize, initializedConnections); + } + } - Future promise = connectTo(entry); - promise.addListener(new FutureListener() { - @Override - public void operationComplete(Future future) throws Exception { - if (future.isSuccess()) { - T conn = future.getNow(); - releaseConnection(entry, conn); - } - releaseConnection(entry); - if (!future.isSuccess()) { - Throwable cause = new RedisConnectionException( - "Can't init enough connections amount! Only " + (minimumIdleSize - initializedConnections.get()) + " from " + minimumIdleSize + " were initialized. Server: " - + entry.getClient().getAddr(), future.cause()); - initPromise.tryFailure(cause); - return; - } + private void createConnection(final boolean checkFreezed, final AtomicInteger requests, final ClientConnectionsEntry entry, final Promise initPromise, + final int minimumIdleSize, final AtomicInteger initializedConnections) { + + if ((checkFreezed && entry.isFreezed()) || !tryAcquireConnection(entry)) { + Throwable cause = new RedisConnectionException( + "Can't init enough connections amount! Only " + (minimumIdleSize - initializedConnections.get()) + " from " + minimumIdleSize + " were initialized. Server: " + + entry.getClient().getAddr()); + initPromise.tryFailure(cause); + return; + } + + Future promise = createConnection(entry); + promise.addListener(new FutureListener() { + @Override + public void operationComplete(Future future) throws Exception { + if (future.isSuccess()) { + T conn = future.getNow(); + releaseConnection(entry, conn); + } + + releaseConnection(entry); + + if (!future.isSuccess()) { + Throwable cause = new RedisConnectionException( + "Can't init enough connections amount! Only " + (minimumIdleSize - initializedConnections.get()) + " from " + minimumIdleSize + " were initialized. Server: " + + entry.getClient().getAddr(), future.cause()); + initPromise.tryFailure(cause); + return; + } - if (initializedConnections.decrementAndGet() == 0) { - initPromise.setSuccess(null); + int value = initializedConnections.decrementAndGet(); + if (value == 0) { + log.info("{} connections initialized for {}", minimumIdleSize, entry.getClient().getAddr()); + initPromise.setSuccess(null); + } else if (value > 0 && !initPromise.isDone()) { + if (requests.incrementAndGet() <= minimumIdleSize) { + createConnection(checkFreezed, requests, entry, initPromise, minimumIdleSize, initializedConnections); } } - }); - } + } + }); } protected abstract int getMinimumIdleSize(ClientConnectionsEntry entry); @@ -137,14 +153,13 @@ abstract class ConnectionPool { } } - StringBuilder errorMsg = new StringBuilder("Connection pool exhausted! All connections are busy. "); - if (!freezed.isEmpty()) { - errorMsg.append(" disconnected hosts: " + freezed); - } + StringBuilder errorMsg = new StringBuilder("Connection pool exhausted! All connections are busy. Try to increase connection pool size."); +// if (!freezed.isEmpty()) { +// errorMsg.append(" Disconnected hosts: " + freezed); +// } if (!zeroConnectionsAmount.isEmpty()) { - errorMsg.append(" hosts with fully busy connections: " + zeroConnectionsAmount); + errorMsg.append(" Hosts with fully busy connections: " + zeroConnectionsAmount); } - errorMsg.append(" Try to increase connection pool size."); RedisConnectionException exception = new RedisConnectionException(errorMsg.toString()); return connectionManager.newFailedFuture(exception); @@ -183,6 +198,10 @@ abstract class ConnectionPool { return promiseSuccessful(entry, conn); } + return createConnection(entry); + } + + private Future createConnection(final ClientConnectionsEntry entry) { final Promise promise = connectionManager.newPromise(); Future connFuture = connect(entry); connFuture.addListener(new FutureListener() {