From e51fd89ccbb290dfbdcb4c42e2f09a1158bbd1cc Mon Sep 17 00:00:00 2001 From: wenshao Date: Tue, 27 Dec 2011 15:52:26 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E6=89=A7=E8=A1=8C=E6=AC=A1?= =?UTF-8?q?=E6=95=B0=E7=9B=91=E6=8E=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../druid/pool/DruidAbstractDataSource.java | 10 ++ .../alibaba/druid/pool/DruidDataSource.java | 2 +- .../druid/pool/PoolablePreparedStatement.java | 56 ++++++----- .../alibaba/druid/pool/PoolableStatement.java | 17 +++- .../ha/balance/LeastConnectionBalancer.java | 34 ++++++- .../pool/ha/balance/RoundRobinBlancer.java | 89 +++++++----------- .../druid/pool/ha/balance/WeightBalancer.java | 94 +++++-------------- 7 files changed, 151 insertions(+), 151 deletions(-) diff --git a/src/main/java/com/alibaba/druid/pool/DruidAbstractDataSource.java b/src/main/java/com/alibaba/druid/pool/DruidAbstractDataSource.java index 80097c21fe..65803f8f76 100644 --- a/src/main/java/com/alibaba/druid/pool/DruidAbstractDataSource.java +++ b/src/main/java/com/alibaba/druid/pool/DruidAbstractDataSource.java @@ -199,6 +199,16 @@ public abstract class DruidAbstractDataSource extends WrapperAdapter implements protected Exception createError = null; + private final AtomicLong executeCount = new AtomicLong(); + + public long getExecuteCount() { + return executeCount.get(); + } + + public void incrementExecuteCount() { + this.executeCount.incrementAndGet(); + } + public Exception getCreateError() { return createError; } diff --git a/src/main/java/com/alibaba/druid/pool/DruidDataSource.java b/src/main/java/com/alibaba/druid/pool/DruidDataSource.java index d2c9be8e63..dcb88db565 100644 --- a/src/main/java/com/alibaba/druid/pool/DruidDataSource.java +++ b/src/main/java/com/alibaba/druid/pool/DruidDataSource.java @@ -1315,7 +1315,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat @Override public String getVersion() { - return VERSION.MajorVersion + "." + VERSION.MinorVersion + "." + VERSION.RevisionVersion + "-2011-12-01 12:13"; + return VERSION.MajorVersion + "." + VERSION.MinorVersion + "." + VERSION.RevisionVersion + "-2011-12-27 14:45"; } } diff --git a/src/main/java/com/alibaba/druid/pool/PoolablePreparedStatement.java b/src/main/java/com/alibaba/druid/pool/PoolablePreparedStatement.java index fb510ac721..eb093de546 100644 --- a/src/main/java/com/alibaba/druid/pool/PoolablePreparedStatement.java +++ b/src/main/java/com/alibaba/druid/pool/PoolablePreparedStatement.java @@ -166,6 +166,7 @@ public class PoolablePreparedStatement extends PoolableStatement implements Prep public ResultSet executeQuery() throws SQLException { checkOpen(); + incrementExecuteCount(); transactionRecord(sql); try { @@ -184,6 +185,7 @@ public class PoolablePreparedStatement extends PoolableStatement implements Prep public int executeUpdate() throws SQLException { checkOpen(); + incrementExecuteCount(); transactionRecord(sql); try { @@ -418,6 +420,7 @@ public class PoolablePreparedStatement extends PoolableStatement implements Prep public boolean execute() throws SQLException { checkOpen(); + incrementExecuteCount(); transactionRecord(sql); try { @@ -439,9 +442,16 @@ public class PoolablePreparedStatement extends PoolableStatement implements Prep } public int[] executeBatch() throws SQLException { + checkOpen(); + + incrementExecuteCount(); transactionRecord(sql); - return super.executeBatch(); + try { + return super.executeBatch(); + } catch (Throwable t) { + throw checkException(t); + } } @Override @@ -802,26 +812,29 @@ public class PoolablePreparedStatement extends PoolableStatement implements Prep public PreparedStatementKey(String sql, String catalog, MethodType methodType) throws SQLException{ this(sql, catalog, methodType, 0, 0, 0, 0, null, null); } - + public PreparedStatementKey(String sql, String catalog, MethodType methodType, int resultSetType, int resultSetConcurrency) throws SQLException{ this(sql, catalog, methodType, resultSetType, resultSetConcurrency, 0, 0, null, null); } - + public PreparedStatementKey(String sql, String catalog, MethodType methodType, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException{ this(sql, catalog, methodType, resultSetType, resultSetConcurrency, resultSetHoldability, 0, null, null); } - - public PreparedStatementKey(String sql, String catalog, MethodType methodType, int autoGeneratedKeys) throws SQLException{ + + public PreparedStatementKey(String sql, String catalog, MethodType methodType, int autoGeneratedKeys) + throws SQLException{ this(sql, catalog, methodType, 0, 0, 0, autoGeneratedKeys, null, null); } - - public PreparedStatementKey(String sql, String catalog, MethodType methodType, int[] columnIndexes) throws SQLException{ + + public PreparedStatementKey(String sql, String catalog, MethodType methodType, int[] columnIndexes) + throws SQLException{ this(sql, catalog, methodType, 0, 0, 0, 0, columnIndexes, null); } - - public PreparedStatementKey(String sql, String catalog, MethodType methodType, String[] columnNames) throws SQLException{ + + public PreparedStatementKey(String sql, String catalog, MethodType methodType, String[] columnNames) + throws SQLException{ this(sql, catalog, methodType, 0, 0, 0, 0, null, columnNames); } @@ -875,51 +888,51 @@ public class PoolablePreparedStatement extends PoolableStatement implements Prep if (this.methodType != that.methodType) { return false; } - + if (this.resultSetType != that.resultSetType) { return false; } - + if (this.resultSetConcurrency != that.resultSetConcurrency) { return false; } - + if (this.resultSetHoldability != that.resultSetHoldability) { return false; } - + if (this.autoGeneratedKeys != that.autoGeneratedKeys) { return false; } - + if (!Arrays.equals(columnIndexes, that.columnIndexes)) { return false; } - + if (!Arrays.equals(columnNames, that.columnNames)) { return false; } return true; } - + public int hashCode() { final int prime = 31; int result = 1; - + result = prime * result + ((sql == null) ? 0 : sql.hashCode()); result = prime * result + ((catalog == null) ? 0 : catalog.hashCode()); result = prime * result + ((methodType == null) ? 0 : methodType.hashCode()); - + result = prime * result + resultSetConcurrency; result = prime * result + resultSetHoldability; result = prime * result + resultSetType; - + result = prime * result + autoGeneratedKeys; - + result = prime * result + Arrays.hashCode(columnIndexes); result = prime * result + Arrays.hashCode(columnNames); - + return result; } @@ -927,6 +940,5 @@ public class PoolablePreparedStatement extends PoolableStatement implements Prep return sql; } - } } diff --git a/src/main/java/com/alibaba/druid/pool/PoolableStatement.java b/src/main/java/com/alibaba/druid/pool/PoolableStatement.java index 0e474b0c44..2f657e1b1c 100644 --- a/src/main/java/com/alibaba/druid/pool/PoolableStatement.java +++ b/src/main/java/com/alibaba/druid/pool/PoolableStatement.java @@ -79,11 +79,16 @@ public class PoolableStatement extends PoolableWrapper implements Statement { } resultSetTrace.clear(); } + + public void incrementExecuteCount() { + this.getPoolableConnection().getConnectionHolder().getDataSource().incrementExecuteCount(); + } @Override public final ResultSet executeQuery(String sql) throws SQLException { checkOpen(); - + + incrementExecuteCount(); transactionRecord(sql); try { @@ -106,6 +111,7 @@ public class PoolableStatement extends PoolableWrapper implements Statement { public final int executeUpdate(String sql) throws SQLException { checkOpen(); + incrementExecuteCount(); transactionRecord(sql); try { @@ -250,6 +256,7 @@ public class PoolableStatement extends PoolableWrapper implements Statement { public final boolean execute(String sql) throws SQLException { checkOpen(); + incrementExecuteCount(); transactionRecord(sql); try { @@ -390,6 +397,8 @@ public class PoolableStatement extends PoolableWrapper implements Statement { @Override public int[] executeBatch() throws SQLException { checkOpen(); + + incrementExecuteCount(); try { return stmt.executeBatch(); @@ -437,6 +446,7 @@ public class PoolableStatement extends PoolableWrapper implements Statement { public final int executeUpdate(String sql, int autoGeneratedKeys) throws SQLException { checkOpen(); + incrementExecuteCount(); transactionRecord(sql); try { @@ -450,6 +460,7 @@ public class PoolableStatement extends PoolableWrapper implements Statement { public final int executeUpdate(String sql, int columnIndexes[]) throws SQLException { checkOpen(); + incrementExecuteCount(); transactionRecord(sql); try { @@ -463,6 +474,7 @@ public class PoolableStatement extends PoolableWrapper implements Statement { public final int executeUpdate(String sql, String columnNames[]) throws SQLException { checkOpen(); + incrementExecuteCount(); transactionRecord(sql); try { @@ -476,6 +488,7 @@ public class PoolableStatement extends PoolableWrapper implements Statement { public final boolean execute(String sql, int autoGeneratedKeys) throws SQLException { checkOpen(); + incrementExecuteCount(); transactionRecord(sql); try { @@ -489,6 +502,7 @@ public class PoolableStatement extends PoolableWrapper implements Statement { public final boolean execute(String sql, int columnIndexes[]) throws SQLException { checkOpen(); + incrementExecuteCount(); transactionRecord(sql); try { @@ -502,6 +516,7 @@ public class PoolableStatement extends PoolableWrapper implements Statement { public final boolean execute(String sql, String columnNames[]) throws SQLException { checkOpen(); + incrementExecuteCount(); transactionRecord(sql); try { diff --git a/src/main/java/com/alibaba/druid/pool/ha/balance/LeastConnectionBalancer.java b/src/main/java/com/alibaba/druid/pool/ha/balance/LeastConnectionBalancer.java index 248801d078..c0bc7367dd 100644 --- a/src/main/java/com/alibaba/druid/pool/ha/balance/LeastConnectionBalancer.java +++ b/src/main/java/com/alibaba/druid/pool/ha/balance/LeastConnectionBalancer.java @@ -3,7 +3,9 @@ package com.alibaba.druid.pool.ha.balance; import java.sql.SQLException; import com.alibaba.druid.pool.ha.DataSourceChangedEvent; +import com.alibaba.druid.pool.ha.DataSourceHolder; import com.alibaba.druid.pool.ha.MultiConnectionHolder; +import com.alibaba.druid.pool.ha.MultiDataSource; import com.alibaba.druid.pool.ha.MultiDataSourceConnection; public class LeastConnectionBalancer extends AbstractBalancer { @@ -14,9 +16,35 @@ public class LeastConnectionBalancer extends AbstractBalancer { } @Override - public MultiConnectionHolder getConnection(MultiDataSourceConnection connectionProxy, String sql) - throws SQLException { - return null; + public MultiConnectionHolder getConnection(MultiDataSourceConnection conn, String sql) throws SQLException { + MultiDataSource multiDataSource = conn.getMultiDataSource(); + + DataSourceHolder dataSource = null; + int leastCount = -1; + + for (DataSourceHolder item : multiDataSource.getDataSources().values()) { + if (!item.isEnable()) { + continue; + } + + int activeCount = item.getDataSource().getActiveCount(); + if (dataSource == null) { + leastCount = activeCount; + dataSource = item; + } else { + if (leastCount > item.getDataSource().getActiveCount()) { + dataSource = item; + leastCount = activeCount; + } + } + } + + if (dataSource == null) { + throw new SQLException("can not get real connection, enableDataSourceCount " + + multiDataSource.getEnabledDataSourceCount()); + } + + return dataSource.getConnection(); } } 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 7b7f88c224..5f3a947855 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 @@ -26,75 +26,56 @@ public class RoundRobinBlancer extends AbstractBalancer { } @Override - public MultiConnectionHolder getConnection(MultiDataSourceConnection connectionProxy, String sql) - throws SQLException { - MultiDataSource multiDataSource = connectionProxy.getMultiDataSource(); + public MultiConnectionHolder getConnection(MultiDataSourceConnection conn, String sql) throws SQLException { + MultiDataSource multiDataSource = conn.getMultiDataSource(); - int tryCount = 0; + int size = multiDataSource.getDataSources().size(); + long connectionId = (int) conn.getId(); - for (;;) { - int size = multiDataSource.getDataSources().size(); - long connectionId = (int) connectionProxy.getId(); + if (size == 0) { + throw new SQLException("can not get connection, no availabe datasources"); + } - if (size == 0) { - throw new SQLException("can not get connection, no availabe datasources"); - } + int index = (int) (connectionId % size); - int index = (int) (connectionId % size); + DataSourceHolder first = null; + DataSourceHolder dataSource = null; - DataSourceHolder first = null; - DataSourceHolder dataSource = null; - - try { - int itemIndex = 0; - for (DataSourceHolder item : multiDataSource.getDataSources().values()) { - if (!item.isEnable()) { - continue; - } - - if (first == null) { - first = item; - } - - if (itemIndex == index) { - dataSource = item; - break; - } - itemIndex++; + try { + int itemIndex = 0; + for (DataSourceHolder item : multiDataSource.getDataSources().values()) { + if (!item.isEnable()) { + continue; } - if (dataSource == null) { - dataSource = first; + if (first == null) { + first = item; } - } catch (Exception ex) { - indexErrorCount.incrementAndGet(); - if (LOG.isDebugEnabled()) { - LOG.debug("getDataSource error, index : " + index, ex); + + if (itemIndex == index) { + dataSource = item; + break; } - continue; + itemIndex++; } if (dataSource == null) { - throw new SQLException("can not get real connection."); + dataSource = first; } - - MultiConnectionHolder conn = null; - - try { - tryCount++; - conn = dataSource.getConnection(); - } catch (SQLException ex) { - LOG.error("getConnection error", ex); - - if (tryCount >= size) { - throw ex; - } - - continue; + } catch (Exception ex) { + indexErrorCount.incrementAndGet(); + if (LOG.isDebugEnabled()) { + LOG.debug("getDataSource error, index : " + index, ex); } - - return conn; } + + if (dataSource == null) { + throw new SQLException("can not get real connection."); + } + + MultiConnectionHolder holder = null; + + return holder; } public long getIndexErrorCount() { 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 34f3e060b3..a50f5c49fd 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 @@ -1,8 +1,6 @@ package com.alibaba.druid.pool.ha.balance; import java.sql.SQLException; -import java.util.concurrent.locks.Condition; -import java.util.concurrent.locks.Lock; import com.alibaba.druid.pool.ha.DataSourceChangedEvent; import com.alibaba.druid.pool.ha.DataSourceHolder; @@ -13,8 +11,7 @@ import com.alibaba.druid.util.ThreadLocalRandom; public class WeightBalancer extends AbstractBalancer { - private int totalWeight = 0; - + private int totalWeight = 0; public void afterDataSourceChanged(DataSourceChangedEvent event) { computeTotalWeight(); @@ -27,7 +24,7 @@ public class WeightBalancer extends AbstractBalancer { return ThreadLocalRandom.current().nextInt(totalWeight); } - + public void computeTotalWeight() { int totalWeight = 0; for (DataSourceHolder holder : getMultiDataSource().getDataSources().values()) { @@ -46,82 +43,39 @@ public class WeightBalancer extends AbstractBalancer { } @Override - public MultiConnectionHolder getConnection(MultiDataSourceConnection connectionProxy, String sql) - throws SQLException { - MultiDataSource multiDataSource = connectionProxy.getMultiDataSource(); - - long maxWaitMillis = multiDataSource.getMaxWaitMillis(); - - long startNano = -1; - if (maxWaitMillis > 0) { - startNano = System.nanoTime(); - } + public MultiConnectionHolder getConnection(MultiDataSourceConnection conn, String sql) throws SQLException { + MultiDataSource multiDataSource = conn.getMultiDataSource(); DataSourceHolder dataSource = null; - final int MAX_RETRY = 10; - for (int i = 0; i < MAX_RETRY; ++i) { - int randomNumber = produceRandomNumber(); - DataSourceHolder first = null; + int randomNumber = produceRandomNumber(); + DataSourceHolder first = null; - boolean needRetry = false; - for (DataSourceHolder item : multiDataSource.getDataSources().values()) { - if (first == null) { - first = item; - } - if (randomNumber >= item.getWeightRegionBegin() && randomNumber < item.getWeightRegionEnd()) { - if (!item.isEnable()) { - needRetry = true; - break; - } - - if (item.getDataSource().isBusy()) { - multiDataSource.incrementBusySkipCount(); - needRetry = true; - break; - } - - dataSource = item; - } - } - - if (needRetry) { - multiDataSource.incrementRetryGetConnectionCount(); + for (DataSourceHolder item : multiDataSource.getDataSources().values()) { + if (!item.isEnable()) { continue; } - if (dataSource == null) { - dataSource = first; + if (first == null) { + first = item; } - if (dataSource == null && i != MAX_RETRY - 1) { - Lock lock = multiDataSource.getLock(); - Condition notFail = multiDataSource.getNotFail(); - lock.lock(); - try { - if (multiDataSource.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(); + if (randomNumber >= item.getWeightRegionBegin() && randomNumber < item.getWeightRegionEnd()) { + if (!item.isEnable()) { + continue; } + + if (item.getDataSource().isBusy()) { + multiDataSource.incrementBusySkipCount(); + break; + } + + dataSource = item; } - break; + } + + if (dataSource == null) { + dataSource = first; } if (dataSource == null) {