diff --git a/src/main/java/com/alibaba/druid/pool/ha/MultiDataSource.java b/src/main/java/com/alibaba/druid/pool/ha/MultiDataSource.java index d1c465ae67..a1337a6ad7 100644 --- a/src/main/java/com/alibaba/druid/pool/ha/MultiDataSource.java +++ b/src/main/java/com/alibaba/druid/pool/ha/MultiDataSource.java @@ -25,6 +25,8 @@ import com.alibaba.druid.filter.Filter; import com.alibaba.druid.logging.Log; import com.alibaba.druid.logging.LogFactory; import com.alibaba.druid.pool.DataSourceAdapter; +import com.alibaba.druid.pool.ha.balance.Balancer; +import com.alibaba.druid.pool.ha.balance.WeightBalancer; import com.alibaba.druid.pool.ha.config.ConfigLoader; import com.alibaba.druid.pool.ha.valid.DataSourceFailureDetecter; import com.alibaba.druid.pool.ha.valid.DefaultDataSourceFailureDetecter; @@ -84,6 +86,8 @@ public class MultiDataSource extends DataSourceAdapter implements MultiDataSourc private long activeCount = 0; private long maxWaitMillis = 0; + private Balancer balancer = new WeightBalancer(); + public long getMaxWaitMillis() { return maxWaitMillis; } @@ -377,82 +381,7 @@ public class MultiDataSource extends DataSourceAdapter implements MultiDataSourc } public MultiConnectionHolder getRealConnection(MultiDataSourceConnection multiConn, String sql) throws SQLException { - - long startNano = -1; - if (maxWaitMillis > 0) { - startNano = System.nanoTime(); - } - - DataSourceHolder dataSource = null; - - final int MAX_RETRY = 10; - for (int i = 0; i < MAX_RETRY; ++i) { - int randomNumber = produceRandomNumber(); - DataSourceHolder first = null; - - boolean needRetry = false; - for (DataSourceHolder item : this.dataSources.values()) { - if (first == null) { - first = item; - } - if (randomNumber >= item.getWeightRegionBegin() && randomNumber < item.getWeightRegionEnd()) { - if (!item.isEnable()) { - needRetry = true; - break; - } - - if (item.getDataSource().isBusy()) { - busySkipCount.incrementAndGet(); - needRetry = true; - break; - } - - dataSource = item; - } - } - - if (needRetry) { - retryGetConnectionCount.incrementAndGet(); - continue; - } - - if (dataSource == null) { - dataSource = first; - } - - if (dataSource == null && i != MAX_RETRY - 1) { - lock.lock(); - try { - if (getEnabledDataSourceCount() == 0) { - try { - if (maxWaitMillis > 0) { - long nano = System.nanoTime() - startNano; - long restNano = maxWaitMillis * 1000 * 1000 - nano; - if (restNano > 0) { - notFail.awaitNanos(restNano); - } else { - break; - } - } else { - notFail.await(); - } - continue; - } catch (InterruptedException e) { - throw new SQLException("interrupted", e); - } - } - } finally { - lock.unlock(); - } - } - break; - } - - if (dataSource == null) { - throw new SQLException("cannot get connection. enabledDataSourceCount " + getEnabledDataSourceCount()); - } - - return dataSource.getConnection(); + return balancer.getConnection(multiConn, sql); } public int getEnabledDataSourceCount() { @@ -471,15 +400,15 @@ public class MultiDataSource extends DataSourceAdapter implements MultiDataSourc public long getRetryGetConnectionCount() { return retryGetConnectionCount.get(); } - + public void incrementRetryGetConnectionCount() { retryGetConnectionCount.incrementAndGet(); } - + public Lock getLock() { return lock; } - + public Condition getNotFail() { return notFail; } @@ -487,7 +416,7 @@ public class MultiDataSource extends DataSourceAdapter implements MultiDataSourc public long getBusySkipCount() { return busySkipCount.get(); } - + public void incrementBusySkipCount() { busySkipCount.incrementAndGet(); } diff --git a/src/main/java/com/alibaba/druid/pool/ha/MultiDataSourceConnection.java b/src/main/java/com/alibaba/druid/pool/ha/MultiDataSourceConnection.java index a87f778567..da01dc119c 100644 --- a/src/main/java/com/alibaba/druid/pool/ha/MultiDataSourceConnection.java +++ b/src/main/java/com/alibaba/druid/pool/ha/MultiDataSourceConnection.java @@ -29,7 +29,7 @@ import com.alibaba.druid.proxy.jdbc.DataSourceProxy; public class MultiDataSourceConnection extends WrapperAdapter implements Connection, ConnectionProxy { - private final MultiDataSource haDataSource; + private final MultiDataSource multiDataSource; private Connection conn; private DataSourceHolder dataSourceHolder; @@ -51,8 +51,8 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect private boolean closed = false; - public MultiDataSourceConnection(MultiDataSource haDataSource, long id){ - this.haDataSource = haDataSource; + public MultiDataSourceConnection(MultiDataSource multiDataSource, long id){ + this.multiDataSource = multiDataSource; this.id = id; } @@ -62,7 +62,7 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect public void checkConnection(String sql) throws SQLException { if (conn == null) { - MultiConnectionHolder connHolder = haDataSource.getRealConnection(this, sql); + MultiConnectionHolder connHolder = multiDataSource.getRealConnection(this, sql); conn = connHolder.getConnection(); this.dataSourceHolder = connHolder.getDataSourceHolder(); } @@ -98,8 +98,8 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect connectedTime = new Date(); } - public MultiDataSource getHaDataSource() { - return haDataSource; + public MultiDataSource getMultiDataSource() { + return multiDataSource; } @Override @@ -156,7 +156,7 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect } this.closed = true; - haDataSource.afterConnectionClosed(this); + multiDataSource.afterConnectionClosed(this); } @Override @@ -329,14 +329,14 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect @Override public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException { - long stmtId = haDataSource.createStatementId(); + long stmtId = multiDataSource.createStatementId(); MultiDataSourceStatement stmt = new MultiDataSourceStatement(this, stmtId, resultSetType, resultSetConcurrency); return stmt; } @Override public Statement createStatement() throws SQLException { - long stmtId = haDataSource.createStatementId(); + long stmtId = multiDataSource.createStatementId(); MultiDataSourceStatement stmt = new MultiDataSourceStatement(this, stmtId); return stmt; } @@ -346,7 +346,7 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect checkConnection(sql); PreparedStatement stmt = conn.prepareStatement(sql); - long stmtId = haDataSource.createStatementId(); + long stmtId = multiDataSource.createStatementId(); return new MultiDataSourcePreparedStatement(this, stmt, sql, stmtId); } @@ -356,7 +356,7 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect checkConnection(sql); CallableStatement stmt = conn.prepareCall(sql); - long stmtId = haDataSource.createStatementId(); + long stmtId = multiDataSource.createStatementId(); return new CallableStatementProxyImpl(this, stmt, sql, stmtId); } @@ -367,7 +367,7 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect checkConnection(sql); PreparedStatement stmt = conn.prepareStatement(sql, resultSetType, resultSetConcurrency); - long stmtId = haDataSource.createStatementId(); + long stmtId = multiDataSource.createStatementId(); return new MultiDataSourcePreparedStatement(this, stmt, sql, stmtId); } @@ -377,7 +377,7 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect checkConnection(sql); CallableStatement stmt = conn.prepareCall(sql, resultSetType, resultSetConcurrency); - long stmtId = haDataSource.createStatementId(); + long stmtId = multiDataSource.createStatementId(); return new CallableStatementProxyImpl(this, stmt, sql, stmtId); } @@ -385,7 +385,7 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect @Override public Statement createStatement(int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException { - long stmtId = haDataSource.createStatementId(); + long stmtId = multiDataSource.createStatementId(); MultiDataSourceStatement stmt = new MultiDataSourceStatement(this, stmtId, resultSetType, resultSetConcurrency, resultSetHoldability); return stmt; @@ -397,7 +397,7 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect checkConnection(sql); PreparedStatement stmt = conn.prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability); - long stmtId = haDataSource.createStatementId(); + long stmtId = multiDataSource.createStatementId(); return new MultiDataSourcePreparedStatement(this, stmt, sql, stmtId); } @@ -408,7 +408,7 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect checkConnection(sql); CallableStatement stmt = conn.prepareCall(sql, resultSetType, resultSetConcurrency, resultSetHoldability); - long stmtId = haDataSource.createStatementId(); + long stmtId = multiDataSource.createStatementId(); return new CallableStatementProxyImpl(this, stmt, sql, stmtId); } @@ -418,7 +418,7 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect checkConnection(sql); PreparedStatement stmt = conn.prepareStatement(sql, autoGeneratedKeys); - long stmtId = haDataSource.createStatementId(); + long stmtId = multiDataSource.createStatementId(); return new MultiDataSourcePreparedStatement(this, stmt, sql, stmtId); } @@ -428,7 +428,7 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect checkConnection(sql); PreparedStatement stmt = conn.prepareStatement(sql, columnIndexes); - long stmtId = haDataSource.createStatementId(); + long stmtId = multiDataSource.createStatementId(); return new MultiDataSourcePreparedStatement(this, stmt, sql, stmtId); } @@ -438,7 +438,7 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect checkConnection(sql); PreparedStatement stmt = conn.prepareStatement(sql, columnNames); - long stmtId = haDataSource.createStatementId(); + long stmtId = multiDataSource.createStatementId(); return new MultiDataSourcePreparedStatement(this, stmt, sql, stmtId); } @@ -578,12 +578,12 @@ public class MultiDataSourceConnection extends WrapperAdapter implements Connect @Override public Properties getProperties() { - return haDataSource.getProperties(); + return multiDataSource.getProperties(); } @Override public DataSourceProxy getDirectDataSource() { - return haDataSource; + return multiDataSource; } @Override diff --git a/src/main/java/com/alibaba/druid/pool/ha/balance/RoundRobinBlancer.java b/src/main/java/com/alibaba/druid/pool/ha/balance/RoundRobinBlancer.java index e298544293..d0178dde98 100644 --- a/src/main/java/com/alibaba/druid/pool/ha/balance/RoundRobinBlancer.java +++ b/src/main/java/com/alibaba/druid/pool/ha/balance/RoundRobinBlancer.java @@ -23,7 +23,7 @@ public class RoundRobinBlancer implements Balancer { @Override public MultiConnectionHolder getConnection(MultiDataSourceConnection connectionProxy, String sql) throws SQLException { - MultiDataSource multiDataSource = connectionProxy.getHaDataSource(); + MultiDataSource multiDataSource = connectionProxy.getMultiDataSource(); int tryCount = 0; diff --git a/src/main/java/com/alibaba/druid/pool/ha/balance/WeightBalancer.java b/src/main/java/com/alibaba/druid/pool/ha/balance/WeightBalancer.java index eac6e071b4..7bf785280d 100644 --- a/src/main/java/com/alibaba/druid/pool/ha/balance/WeightBalancer.java +++ b/src/main/java/com/alibaba/druid/pool/ha/balance/WeightBalancer.java @@ -14,9 +14,9 @@ public class WeightBalancer implements Balancer { @Override public MultiConnectionHolder getConnection(MultiDataSourceConnection connectionProxy, String sql) throws SQLException { - MultiDataSource haDataSource = connectionProxy.getHaDataSource(); + MultiDataSource multiDataSource = connectionProxy.getMultiDataSource(); - long maxWaitMillis = haDataSource.getMaxWaitMillis(); + long maxWaitMillis = multiDataSource.getMaxWaitMillis(); long startNano = -1; if (maxWaitMillis > 0) { @@ -27,11 +27,11 @@ public class WeightBalancer implements Balancer { final int MAX_RETRY = 10; for (int i = 0; i < MAX_RETRY; ++i) { - int randomNumber = haDataSource.produceRandomNumber(); + int randomNumber = multiDataSource.produceRandomNumber(); DataSourceHolder first = null; boolean needRetry = false; - for (DataSourceHolder item : haDataSource.getDataSources().values()) { + for (DataSourceHolder item : multiDataSource.getDataSources().values()) { if (first == null) { first = item; } @@ -42,7 +42,7 @@ public class WeightBalancer implements Balancer { } if (item.getDataSource().isBusy()) { - haDataSource.incrementBusySkipCount(); + multiDataSource.incrementBusySkipCount(); needRetry = true; break; } @@ -52,7 +52,7 @@ public class WeightBalancer implements Balancer { } if (needRetry) { - haDataSource.incrementRetryGetConnectionCount(); + multiDataSource.incrementRetryGetConnectionCount(); continue; } @@ -61,11 +61,11 @@ public class WeightBalancer implements Balancer { } if (dataSource == null && i != MAX_RETRY - 1) { - Lock lock = haDataSource.getLock(); - Condition notFail = haDataSource.getNotFail(); + Lock lock = multiDataSource.getLock(); + Condition notFail = multiDataSource.getNotFail(); lock.lock(); try { - if (haDataSource.getEnabledDataSourceCount() == 0) { + if (multiDataSource.getEnabledDataSourceCount() == 0) { try { if (maxWaitMillis > 0) { long nano = System.nanoTime() - startNano; @@ -91,7 +91,7 @@ public class WeightBalancer implements Balancer { } if (dataSource == null) { - throw new SQLException("cannot get connection. enabledDataSourceCount " + haDataSource.getEnabledDataSourceCount()); + throw new SQLException("cannot get connection. enabledDataSourceCount " + multiDataSource.getEnabledDataSourceCount()); } return dataSource.getConnection();